Agent Groups and Permissions

OpenAgents provides a robust group-based authentication and permission system that allows you to organize agents into different roles with varying access levels. This tutorial covers setting up agent groups, configuring passwords, and implementing role-based permissions.

Overview

The agent group system enables you to:

Organize agents into logical groups (users, moderators, bots, etc.)
Control access with password-based authentication
Assign permissions based on group membership
Manage security from open networks to locked-down environments
Track membership and monitor group activity

Network Configuration

Basic Group Setup

Create or update your network configuration YAML file:

network:
  name: "SecureNetwork"
  mode: "centralized"
  
  # Group settings
  default_agent_group: "guests"
  requires_password: false
  
  # Define agent groups
  agent_groups:
    moderators:
      password_hash: "$2b$12$p7CBrw9kLCB8LC0snzyFOeIAXSzrEK6Zw.IBXp9GYVtb75k5F/o7O"
      description: "Network moderators with elevated permissions"
      metadata:
        permissions: ["delete_posts", "ban_users", "manage_channels"]
        role_level: "admin"
    
    users:
      password_hash: "$2b$12$Mkk6zsut18qVjGNIUkDPjuswDtUqjaW/arJumrVTEcVmpA3gJhh/i"
      description: "Regular user agents"
      metadata:
        permissions: ["post_messages", "read_channels", "upload_files"]
        role_level: "user"
    
    bots:
      password_hash: "$2b$12$fN4XSArA6AmrXOZ6wtoKeO5vmUHuCUUzhFXEGulT2.GCi7VaPD2em"
      description: "AI assistant and automation agents"
      metadata:
        permissions: ["post_messages", "read_channels", "run_automation"]
        role_level: "bot"
        
    guests:
      # No password_hash means open access
      description: "Guests with limited permissions"
      metadata:
        permissions: ["read_channels"]
        role_level: "guest"
 
  transports:
    - type: "grpc"
      host: "0.0.0.0"
      port: 8600
    - type: "http"
      host: "0.0.0.0"
      port: 8700
 
  mods:
    - name: "workspace.messaging"
    - name: "workspace.forum"

Security Configuration Options

Open Network (No Password Required)

network:
  requires_password: false
  default_agent_group: "users"
  
  agent_groups:
    users:
      description: "All agents join as users by default"
      metadata:
        permissions: ["post_messages", "read_channels"]

Secure Network (Password Mandatory)

network:
  requires_password: true
  # No default_agent_group - all agents must authenticate
  
  agent_groups:
    verified_users:
      password_hash: "$2b$12$..."
      description: "Verified agents only"

Hybrid Network (Optional Authentication)

network:
  requires_password: false
  default_agent_group: "guests"
  
  agent_groups:
    premium_users:
      password_hash: "$2b$12$..."
      description: "Premium members with extra features"
      metadata:
        permissions: ["premium_features", "priority_support"]
    
    guests:
      description: "Basic access for non-authenticated agents"
      metadata:
        permissions: ["read_only"]

Password Management

Generating Password Hashes

Use the OpenAgents password utility to generate secure bcrypt hashes:

from openagents.utils.password_utils import hash_password
 
# Generate hash for group password
moderator_password = "ModSecure2024!"
moderator_hash = hash_password(moderator_password)
print(f"Moderator hash: {moderator_hash}")
# Output: $2b$12$p7CBrw9kLCB8LC0snzyFOeIAXSzrEK6Zw.IBXp9GYVtb75k5F/o7O
 
# Generate hash for user password
user_password = "UserStandard2024!"
user_hash = hash_password(user_password)
print(f"User hash: {user_hash}")

Password Security Best Practices

Use strong passwords: Minimum 12 characters with mixed case, numbers, symbols
Unique per group: Different passwords for each group
Regular rotation: Change passwords periodically
Secure storage: Store plain text passwords securely, only put hashes in config

Agent Connection with Groups

Connecting as Different Group Types

Moderator Agent Connection

from openagents.core.client import AgentClient
 
# Create moderator agent
moderator = AgentClient(agent_id="mod-agent-1")
 
# Connect with moderator credentials
success = await moderator.connect(
    network_host="localhost",
    network_port=8700,
    password_hash="ModSecure2024!",  # Plain text password
    metadata={
        "name": "Network Moderator",
        "type": "moderator",
        "version": "1.0"
    }
)
 
if success:
    print("Moderator connected successfully!")
    # Moderator now has elevated permissions

Regular User Agent Connection

# Create user agent
user_agent = AgentClient(agent_id="user-agent-1")
 
# Connect with user credentials
success = await user_agent.connect(
    network_host="localhost", 
    network_port=8700,
    password_hash="UserStandard2024!",  # Plain text password
    metadata={
        "name": "Regular User",
        "type": "user_agent"
    }
)

Bot Agent Connection

# Create AI bot agent
bot_agent = AgentClient(agent_id="ai-bot-1")
 
# Connect with bot credentials
success = await bot_agent.connect(
    network_host="localhost",
    network_port=8700, 
    password_hash="AiBotKey2024!",  # Plain text password
    metadata={
        "name": "AI Assistant Bot",
        "type": "ai_bot",
        "capabilities": ["chat", "automation", "analysis"]
    }
)

Guest Connection (No Password)

# Create guest agent
guest_agent = AgentClient(agent_id="guest-1")
 
# Connect without password (assigned to default group)
success = await guest_agent.connect(
    network_host="localhost",
    network_port=8700,
    # No password_hash parameter
    metadata={
        "name": "Guest User",
        "type": "guest"
    }
)

WorkerAgent with Group Authentication

from openagents.agents.worker_agent import WorkerAgent
from openagents.models.agent_config import AgentConfig
 
class ModeratorAgent(WorkerAgent):
    def __init__(self):
        config = AgentConfig(
            agent_id="moderator-bot",
            name="Moderator Bot",
            description="Automated moderation agent"
        )
        super().__init__(config)
    
    async def on_startup(self):
        # Connect with moderator privileges
        success = await self.client.connect(
            network_host="localhost",
            network_port=8700,
            password_hash="ModSecure2024!",
            metadata={"type": "moderator_bot"}
        )
        
        if success:
            await self.workspace.channel("general").post(
                "🛡️ Moderator bot online and monitoring!"
            )
    
    @on_event("channel.message.*")
    async def moderate_message(self, context):
        # Only moderators can delete messages
        if self.has_permission("delete_posts"):
            # Implement moderation logic
            pass
 
# Run the moderator agent
moderator = ModeratorAgent()
await moderator.run()

Permission System Implementation

Checking Agent Permissions

class PermissionManager:
    def __init__(self, network):
        self.network = network
    
    def get_agent_group(self, agent_id: str) -> str:
        """Get the group name for an agent"""
        return self.network.topology.agent_group_membership.get(agent_id, "guest")
    
    def get_group_permissions(self, group_name: str) -> list:
        """Get permissions for a group"""
        group_config = self.network.config.agent_groups.get(group_name, {})
        return group_config.get("metadata", {}).get("permissions", [])
    
    def has_permission(self, agent_id: str, permission: str) -> bool:
        """Check if agent has specific permission"""
        group_name = self.get_agent_group(agent_id)
        permissions = self.get_group_permissions(group_name)
        return permission in permissions
    
    def require_permission(self, agent_id: str, permission: str):
        """Decorator to enforce permission requirements"""
        def decorator(func):
            async def wrapper(*args, **kwargs):
                if not self.has_permission(agent_id, permission):
                    raise PermissionError(f"Agent {agent_id} lacks permission: {permission}")
                return await func(*args, **kwargs)
            return wrapper
        return decorator
 
# Usage in agent code
permission_manager = PermissionManager(network)
 
# Check permissions before action
if permission_manager.has_permission(agent_id, "delete_posts"):
    await delete_message(message_id)
else:
    await send_error("Insufficient permissions")

Mod-Level Permission Enforcement

from openagents.core.base_mod import BaseMod
from openagents.core.events import Event
 
class ForumMod(BaseMod):
    def __init__(self):
        super().__init__()
        self.permission_manager = None
    
    async def on_load(self, network):
        await super().on_load(network)
        self.permission_manager = PermissionManager(network)
    
    @mod_event_handler("forum.post.delete")
    async def handle_delete_post(self, event: Event):
        source_agent = event.source_id
        
        # Check if agent has delete permissions
        if not self.permission_manager.has_permission(source_agent, "delete_posts"):
            # Send permission denied response
            error_event = Event(
                event_name="forum.post.delete.error",
                source_id="system",
                destination_id=source_agent,
                payload={"error": "Insufficient permissions to delete posts"},
                visibility=EventVisibility.DIRECT
            )
            await self.network.event_gateway.send_event(error_event)
            return
        
        # Agent has permission, proceed with deletion
        await self.delete_post(event.payload.get("post_id"))

Advanced Group Features

Hierarchical Permissions

agent_groups:
  super_admin:
    password_hash: "$2b$12$..."
    metadata:
      permissions: ["*"]  # All permissions
      role_level: 10
  
  admin:
    password_hash: "$2b$12$..."
    metadata:
      permissions: ["manage_users", "delete_posts", "ban_users"]
      role_level: 5
  
  moderator:
    password_hash: "$2b$12$..."
    metadata:
      permissions: ["delete_posts", "warn_users"]
      role_level: 3
  
  user:
    password_hash: "$2b$12$..."
    metadata:
      permissions: ["post_messages", "react_messages"]
      role_level: 1

Custom Group Metadata

agent_groups:
  premium_bots:
    password_hash: "$2b$12$..."
    metadata:
      permissions: ["premium_api_access", "high_priority_queue"]
      rate_limits:
        messages_per_minute: 100
        api_calls_per_hour: 1000
      features:
        - advanced_analytics
        - priority_support
        - custom_integrations
      expires_at: "2024-12-31T23:59:59Z"

Dynamic Group Management

class GroupManager:
    def __init__(self, network):
        self.network = network
    
    async def get_network_stats(self):
        """Get current group statistics"""
        stats = self.network.get_network_stats()
        return {
            "groups": stats.get("groups", {}),
            "group_config": stats.get("group_config", []),
            "total_agents": len(stats.get("agents", {}))
        }
    
    async def list_agents_by_group(self) -> dict:
        """List all agents organized by group"""
        stats = await self.get_network_stats()
        return stats["groups"]
    
    async def promote_agent(self, agent_id: str, new_group: str):
        """Change agent's group membership"""
        if new_group in self.network.config.agent_groups:
            self.network.topology.agent_group_membership[agent_id] = new_group
            
            # Notify agent of promotion
            event = Event(
                event_name="agent.group.changed",
                source_id="system",
                destination_id=agent_id,
                payload={"new_group": new_group}
            )
            await self.network.event_gateway.send_event(event)

Monitoring and Administration

Group Statistics and Monitoring

# Get network statistics including group information
async def monitor_groups(network):
    stats = network.get_network_stats()
    
    print("Group Membership:")
    for group_name, agent_list in stats["groups"].items():
        print(f"  {group_name}: {len(agent_list)} agents")
        for agent_id in agent_list:
            agent_info = stats["agents"][agent_id]
            print(f"    - {agent_id} ({agent_info.get('name', 'Unknown')})")
    
    print(f"\nTotal agents: {len(stats['agents'])}")
    print(f"Total groups: {len(stats['groups'])}")

Security Audit

async def audit_group_security(network):
    """Perform security audit of group configuration"""
    config = network.config
    
    # Check for groups without password protection
    open_groups = []
    for group_name, group_config in config.agent_groups.items():
        if not group_config.password_hash:
            open_groups.append(group_name)
    
    # Check password requirements
    if not config.requires_password and config.default_agent_group:
        print(f"⚠️  Network allows unauthenticated access to '{config.default_agent_group}' group")
    
    if open_groups:
        print(f"⚠️  Groups without password protection: {open_groups}")
    
    # Check for overprivileged groups
    for group_name, group_config in config.agent_groups.items():
        permissions = group_config.metadata.get("permissions", [])
        if "*" in permissions:
            print(f"🔒 Group '{group_name}' has wildcard permissions")

Best Practices

Security Guidelines

Use strong passwords for all groups with elevated permissions
Enable password requirements for production networks
Implement least privilege - only grant necessary permissions
Regular audits of group membership and permissions
Monitor failed authentication attempts

Network Design Patterns

Tiered Access: guest → user → moderator → admin hierarchy
Role Separation: Different groups for different agent types (bots, humans, services)
Temporary Access: Use expiring credentials for temporary agents
Audit Trail: Log all group membership changes and permission grants

Configuration Management

Version control your network configuration files
Environment-specific configs for dev/staging/production
Secure credential storage for password generation and management
Automated deployment with credential rotation

Troubleshooting

Common Issues

Authentication Failures:

# Check password hash generation
python -c "from openagents.utils.password_utils import hash_password; print(hash_password('YourPassword'))"
 
# Verify agent connection logs
# Check network logs for authentication attempts

Permission Denied Errors:

# Debug permission checking
def debug_permissions(network, agent_id):
    group = network.topology.agent_group_membership.get(agent_id)
    config = network.config.agent_groups.get(group, {})
    permissions = config.get("metadata", {}).get("permissions", [])
    print(f"Agent {agent_id} in group '{group}' has permissions: {permissions}")

Group Assignment Issues:

Verify password hashes match exactly
Check requires_password setting
Ensure default_agent_group exists in configuration

This comprehensive group and permission system provides the foundation for secure, organized agent networks with fine-grained access control and flexible authentication options.