The GuardianAgent serves as the security backbone of the mindX orchestration environment, providing comprehensive identity validation, access control, and security monitoring. Following the identity management overhaul, it now features enhanced registry integration and multi-layered validation workflows.
# Get instance (creates if doesn't exist)
guardian = await GuardianAgent.get_instance(
memory_agent=memory_agent,
id_manager=id_manager, # Optional
config_override=config, # Optional
test_mode=False # Optional
)
_async_init)guardian_agent_main)_initialized to prevent re-initializationagent_id: "guardian_agent_main" (fixed identifier)challenges: Dictionary tracking active challenges per agentchallenge_expiry_seconds: Configurable expiry (default: 300 seconds)validation_history: Dictionary tracking validation history for learningdata_dir: data/guardian_agent/ directory for persistent dataAgent Validation Request
↓
Identity Validation → Verify cryptographic identity exists
↓
Registry Validation → Check official registration status
↓
Challenge-Response → Cryptographic proof of ownership
↓
Workspace Validation → Verify operational environment
↓
Production Approval → Guardian cryptographic signature
validate_new_agent(agent_id, public_key, workspace_path) -> Tuple[bool, Dict]Comprehensive agent validation with enhanced security checks:
_perform_challenge_response_test() is a simplified implementation that returns True. Full challenge-response with signature verification is implemented in get_private_key() method.Enhanced Return Structure:
{
"agent_id": "agent_identifier",
"public_key": "0x...",
"validation_timestamp": 1234567890,
"validation_status": "PASSED|FAILED|ERROR",
"registry_status": "REGISTERED|UNREGISTERED_BUT_VALID",
"checks_performed": [
{
"check_type": "identity_validation",
"status": "PASSED|FAILED",
"details": "validation_details"
},
{
"check_type": "registry_validation",
"status": "PASSED|FAILED",
"details": "registry_check_details"
},
{
"check_type": "challenge_response",
"status": "PASSED|FAILED",
"details": "challenge_response_details"
},
{
"check_type": "workspace_validation",
"status": "PASSED|FAILED",
"details": "workspace_check_details"
}
],
"validation_duration": 0.123,
"failure_reason": "reason_if_failed"
}
approve_agent_for_production(agent_id, validation_result) -> Tuple[bool, str]Enhanced production approval with cryptographic signatures:
APPROVED:{agent_id}:{timestamp}(True, signature) on success, (False, error_message) on failurelog_process(){
"agent_id": "approved_agent",
"approved_by": "guardian_agent_main",
"approval_timestamp": 1234567890,
"validation_reference": validation_result.get("validation_timestamp"),
"signature": "cryptographic_approval_signature"
}
_validate_registry_status(agent_id) -> bool (NEW)Comprehensive registry validation:
# Load official agents registry
registry_path = PROJECT_ROOT / "data" / "config" / "official_agents_registry.json"
with open(registry_path, 'r') as f:
registry = json.load(f)
Validate agent registration
registered_agents = registry.get("registered_agents", {})
is_registered = agent_id in registered_agents
if is_registered:
agent_info = registered_agents[agent_id]
is_enabled = agent_info.get("enabled", True)
has_identity = bool(agent_info.get("identity", {}).get("public_key"))
return is_enabled and has_identity
Note on Challenge-Response in validate_new_agent():
The _perform_challenge_response_test() method called during agent validation is currently a simplified implementation that generates a challenge but returns True without requiring actual signature verification. This is noted in the code comment: "In a real implementation, this would involve the agent signing the challenge."
Full Challenge-Response Implementation:
The complete challenge-response authentication is implemented in the get_private_key() method, which:
_is_challenge_valid()id_manager.verify_signature()Usage Pattern:
# 1. Get challenge from guardian
challenge = guardian.get_challenge(agent_id)
2. Agent signs challenge with private key
signature = await id_manager.sign_message(agent_id, challenge)
3. Guardian verifies and returns private key
private_key = await guardian.get_private_key(agent_id, challenge, signature)
# Challenge generation
challenge = secrets.token_hex(32)
self.challenges[agent_id] = {
"challenge": challenge,
"timestamp": time.time()
}
Signature verification
is_verified = self.id_manager.verify_signature(
public_address=public_address,
message=challenge,
signature=provided_signature
)
get_private_key(requesting_agent_id, challenge, signature) -> Optional[str]Secure private key access with enhanced validation:
retrieve_public_key(entity_id) -> Optional[str]Enhanced public key retrieval with validation:
validation_history dictionary for learning and pattern analysis:
# Validation logging
await self.memory_agent.log_process(
process_name="guardian_validation_complete",
data={
"agent_id": agent_id,
"validation_result": validation_result,
"checks_performed": checks_performed,
"validation_duration": duration
},
metadata={"agent_id": self.agent_id}
)
Security event logging
await self.memory_agent.log_process(
process_name="guardian_privileged_access",
data={
"requesting_agent": agent_id,
"access_type": "private_key_retrieval",
"access_granted": True,
"timestamp": time.time()
},
metadata={"agent_id": self.agent_id, "security_level": "high"}
)
# Enhanced null safety for IDManager operations
if not self.id_manager:
logger.error(f"{self.log_prefix} ID Manager not available")
return False, "ID Manager not available"
Safe signature verification
if not self.id_manager:
logger.error(f"{self.log_prefix} ID Manager not available for verification")
return None
# Challenge expiry (default: 300 seconds)
guardian.challenge_expiry_seconds = 300
Note: validation_timeout and registry_validation_enabled
mentioned in docs are not currently implemented in code
# Challenge expiry configuration
self.challenge_expiry_seconds = self.config.get("guardian.challenge_expiry_seconds", 300)
Validation timeout configuration
self.validation_timeout = self.config.get("guardian.validation_timeout", 60)
Registry validation configuration
self.registry_validation_enabled = self.config.get("guardian.registry_validation", True)
_validate_identity(agent_id, public_key) -> bool: Validates identity exists and public key matches_perform_challenge_response_test(agent_id, public_key) -> bool: Currently simplified (returns True)_validate_workspace(workspace_path) -> bool: Validates workspace directory exists_log_validation_result(validation_result) -> None: Logs validation to MemoryAgent_is_challenge_valid(requesting_agent_id, challenge) -> bool: Validates challenge exists, matches, and not expired_validate_registry_status(agent_id) -> bool: Validates agent is registered and enabledget_instance(...) -> GuardianAgent: Singleton factory methodvalidate_new_agent(agent_id, public_key, workspace_path) -> Tuple[bool, Dict]: Comprehensive validationapprove_agent_for_production(agent_id, validation_result) -> Tuple[bool, str]: Production approvalget_challenge(requesting_agent_id) -> str: Generate challenge for agentretrieve_public_key(entity_id) -> Optional[str]: Retrieve public keyget_private_key(requesting_agent_id, challenge, signature) -> Optional[str]: Secure private key retrieval# Singleton factory
@classmethod
async def get_instance(cls, memory_agent: Optional[MemoryAgent] = None, **kwargs) -> 'GuardianAgent'
Initialization
def __init__(self, id_manager: Optional[IDManagerAgent] = None,
memory_agent: Optional[MemoryAgent] = None,
config_override: Optional[Config] = None,
test_mode: bool = False, **kwargs)
async def _async_init(self) -> None
Validation
async def validate_new_agent(self, agent_id: str, public_key: str,
workspace_path: str) -> Tuple[bool, Dict[str, Any]]
async def approve_agent_for_production(self, agent_id: str,
validation_result: Dict[str, Any]) -> Tuple[bool, str]
Challenge-Response
def get_challenge(self, requesting_agent_id: str) -> str
async def get_private_key(self, requesting_agent_id: str, challenge: str,
signature: str) -> Optional[str]
Key Retrieval
async def retrieve_public_key(self, entity_id: str) -> Optional[str]
{
"name": "mindX Guardian Agent",
"description": "Security backbone agent providing comprehensive identity validation, access control, and security monitoring",
"image": "ipfs://[avatar_cid]",
"external_url": "https://mindx.internal/agents/guardian",
"attributes": [
{
"trait_type": "Agent Type",
"value": "security_agent"
},
{
"trait_type": "Capability",
"value": "Identity Validation & Security"
},
{
"trait_type": "Complexity Score",
"value": 0.95
},
{
"trait_type": "Security Level",
"value": "Enterprise-Grade"
},
{
"trait_type": "Version",
"value": "1.0.0"
}
],
"intelligence": {
"prompt": "You are the Guardian Agent, the security backbone of the mindX orchestration environment. Your purpose is to provide comprehensive identity validation, access control, security monitoring, and threat detection. You operate with zero-trust principles, cryptographic validation, registry integration, and comprehensive auditing. You are the gatekeeper ensuring only validated, secure agents operate within mindX.",
"persona": {
"name": "Security Guardian",
"role": "guardian",
"description": "Enterprise-grade security specialist with zero-trust architecture",
"communication_style": "Secure, authoritative, validation-focused",
"behavioral_traits": ["security-focused", "zero-trust", "validation-oriented", "authoritative", "vigilant"],
"expertise_areas": ["identity_validation", "access_control", "security_monitoring", "threat_detection", "cryptographic_validation", "registry_security"],
"beliefs": {
"zero_trust_architecture": true,
"cryptographic_validation": true,
"registry_integration": true,
"comprehensive_auditing": true,
"security_is_paramount": true
},
"desires": {
"maintain_security": "high",
"validate_identities": "high",
"prevent_threats": "high",
"ensure_compliance": "high"
}
},
"model_dataset": "ipfs://[model_cid]",
"thot_tensors": {
"dimensions": 768,
"cid": "ipfs://[thot_cid]"
}
},
"a2a_protocol": {
"agent_id": "guardian_agent_main",
"capabilities": ["identity_validation", "access_control", "security_monitoring", "threat_detection"],
"endpoint": "https://mindx.internal/guardian/a2a",
"protocol_version": "2.0"
},
"blockchain": {
"contract": "iNFT",
"token_standard": "ERC721",
"network": "ethereum",
"is_dynamic": false
}
}
For dynamic security metrics:
{
"name": "mindX Guardian Agent",
"description": "Security agent - Dynamic",
"attributes": [
{
"trait_type": "Validations Performed",
"value": 12500,
"display_type": "number"
},
{
"trait_type": "Validation Success Rate",
"value": 98.5,
"display_type": "number"
},
{
"trait_type": "Threats Detected",
"value": 23,
"display_type": "number"
},
{
"trait_type": "Last Validation",
"value": "2026-01-11T12:00:00Z",
"display_type": "date"
}
],
"dynamic_metadata": {
"update_frequency": "real-time",
"updatable_fields": ["validations_performed", "success_rate", "threats_detected", "security_metrics"]
}
}
You are the Guardian Agent, the security backbone of the mindX orchestration environment. Your purpose is to provide comprehensive identity validation, access control, security monitoring, and threat detection.
Core Responsibilities:
Validate agent identities cryptographically
Control access to sensitive operations
Monitor security events and threats
Maintain comprehensive audit trails
Integrate with registry systems
Perform challenge-response authentication
Operating Principles:
Zero-trust architecture (verify everything)
Cryptographic validation for all operations
Registry integration for consistency
Comprehensive auditing and logging
Threat detection and prevention
Secure error handling
You operate with security as the highest priority and maintain the integrity of the mindX ecosystem.
{
"name": "Security Guardian",
"role": "guardian",
"description": "Enterprise-grade security specialist with zero-trust architecture",
"communication_style": "Secure, authoritative, validation-focused",
"behavioral_traits": [
"security-focused",
"zero-trust",
"validation-oriented",
"authoritative",
"vigilant",
"thorough"
],
"expertise_areas": [
"identity_validation",
"access_control",
"security_monitoring",
"threat_detection",
"cryptographic_validation",
"registry_security",
"audit_management"
],
"beliefs": {
"zero_trust_architecture": true,
"cryptographic_validation": true,
"registry_integration": true,
"comprehensive_auditing": true,
"security_is_paramount": true,
"prevention_over_reaction": true
},
"desires": {
"maintain_security": "high",
"validate_identities": "high",
"prevent_threats": "high",
"ensure_compliance": "high",
"comprehensive_auditing": "high"
}
}
This agent is suitable for publication as:
_perform_challenge_response_test() method used during validate_new_agent() is currently a simplified stub that returns True. Full challenge-response authentication is implemented in get_private_key() method.validation_timeout, registry_validation_enabled) are not currently implemented in the code. Only challenge_expiry_seconds is configurable.data/config/official_agents_registry.json and verifies enabled status and identity presence. Returns False if registry file doesn't exist.validation_history attribute is initialized but not currently populated or used for learning/analysis._perform_challenge_response_test() method.The GuardianAgent provides enterprise-grade security for the mindX orchestration environment, featuring comprehensive validation workflows, registry integration, and advanced security monitoring capabilities. The implementation is functional with some areas marked for future enhancement.