"""
Agent Deletion Validation Module
Provides comprehensive validation for agent deletion operations with security
checks, permission verification, and state validation.
Architecture Integration:
- Design Patterns: Decorator pattern for validation layers, Strategy for validation rules
- Security Model: Defense-in-depth with permission and state validation
- Performance Profile: O(1) validation checks with minimal overhead
Technical Decisions:
- Validation Layers: Format validation → Permission validation → State validation
- Security-First: All deletion requests validated against security policies
- Audit Integration: Complete audit trail for all validation decisions
Dependencies & Integration:
- Internal: Type system, security contracts, audit logging
- Validation Framework: Reusable validation decorators and utilities
Quality Assurance:
- Test Coverage: Property-based testing for all validation rules
- Error Handling: Clear validation errors with actionable messages
- Contract Validation: All operations protected by security contracts
Author: ADDER_4 | Created: 2025-06-26 | Last Modified: 2025-06-26
"""
import re
from datetime import datetime
from functools import wraps
from typing import Any, Callable, Optional
# Import audit logging
from src.boundaries.audit import AuditCategory, AuditLevel, get_audit_logger
# Import security contracts
from src.contracts.security import SecurityError
from src.models.agent import AgentState
# Import type system
from src.models.ids import AgentId
from src.models.security import SecurityContext, SecurityLevel
from src.models.validation import ValidationError
from src.utils.contracts_shim import ensure, require
# Agent name validation pattern
AGENT_NAME_PATTERN = re.compile(r"^Agent_\d+$")
class AgentDeletionValidator:
"""
Comprehensive validator for agent deletion operations.
Enforces security policies, validates permissions, and ensures
safe deletion of agents with complete audit trails.
Contracts:
Preconditions:
- Validator must be initialized before use
- Security context must be established
Postconditions:
- All validation decisions are logged
- Clear errors returned for validation failures
Invariants:
- Validation rules are consistently applied
- Audit trails are complete for all validations
"""
def __init__(self):
"""Initialize agent deletion validator."""
self._audit_logger = None
self._validation_stats = {
"total_validations": 0,
"successful_validations": 0,
"failed_validations": 0,
"permission_denials": 0,
"state_conflicts": 0,
}
async def initialize(self) -> None:
"""Initialize validator with audit logging."""
self._audit_logger = get_audit_logger()
def validate_agent_name_format(self, agent_name: str) -> bool:
"""
Validate agent name follows required format.
Args:
agent_name: Agent name to validate
Returns:
bool: True if valid format, False otherwise
"""
if not agent_name or not isinstance(agent_name, str):
return False
return bool(AGENT_NAME_PATTERN.match(agent_name))
async def validate_deletion_permissions(
self,
agent_id: AgentId,
session_id: str,
security_context: SecurityContext,
force: bool = False,
) -> bool:
"""
Validate permissions for agent deletion.
Checks if the requesting session has permission to delete
the specified agent based on ownership and security policies.
Args:
agent_id: ID of agent to delete
session_id: ID of requesting session
security_context: Security context of request
force: Whether force deletion is requested
Returns:
bool: True if permitted, raises SecurityError otherwise
Raises:
SecurityError: Insufficient permissions for deletion
"""
# TODO: Implement actual permission checking based on:
# - Session ownership of agent
# - Security level requirements
# - Force deletion permissions
# - Cross-session deletion policies
# For now, basic ownership check
# In real implementation, would check against security policies
if force and security_context.security_level != SecurityLevel.ADMIN:
self._validation_stats["permission_denials"] += 1
if self._audit_logger:
await self._audit_logger.log_event(
level=AuditLevel.WARNING,
category=AuditCategory.SECURITY,
operation="agent_deletion_permission_denied",
resource_type="agent",
resource_id=str(agent_id),
success=False,
error_message="Force deletion requires admin privileges",
metadata={
"session_id": session_id,
"force_requested": True,
"security_level": security_context.security_level.value,
},
)
raise SecurityError(
"Force deletion requires admin privileges. "
"Current security level: " + security_context.security_level.value
)
return True
async def validate_agent_state_for_deletion(
self, agent_state: AgentState, force: bool = False
) -> bool:
"""
Validate agent state allows deletion.
Checks for active operations, critical state, and other
conditions that might prevent safe deletion.
Args:
agent_state: Current agent state
force: Whether to override state checks
Returns:
bool: True if deletion allowed, raises error otherwise
Raises:
ValidationError: Agent state prevents deletion
"""
if not force:
# Check for critical operations
if agent_state.has_critical_operations:
critical_ops = agent_state.get_critical_operations()
self._validation_stats["state_conflicts"] += 1
raise ValidationError(
f"Agent has {len(critical_ops)} critical operations in progress. "
f"Operations: {', '.join(op.operation_type for op in critical_ops)}. "
"Use force=True to override."
)
# Check for active tasks
if agent_state.active_task_count > 0:
raise ValidationError(
f"Agent has {agent_state.active_task_count} active tasks. "
"Complete or cancel tasks before deletion, or use force=True."
)
# Check for uncommitted changes
if agent_state.has_uncommitted_changes:
raise ValidationError(
"Agent has uncommitted changes in working directory. "
"Commit or discard changes before deletion, or use force=True."
)
return True
async def validate_deletion_request(
self,
agent_name: str,
agent_id: Optional[AgentId],
agent_state: Optional[AgentState],
session_id: str,
security_context: SecurityContext,
force: bool = False,
) -> bool:
"""
Comprehensive validation of agent deletion request.
Performs all validation checks in sequence and logs results.
Args:
agent_name: Name of agent to delete
agent_id: ID of agent (if found)
agent_state: Current agent state (if available)
session_id: Requesting session ID
security_context: Security context
force: Whether force deletion requested
Returns:
bool: True if all validations pass
Raises:
ValidationError: Validation failure
SecurityError: Security policy violation
"""
self._validation_stats["total_validations"] += 1
try:
# Step 1: Validate agent name format
if not self.validate_agent_name_format(agent_name):
raise ValidationError(
f"Invalid agent name format. Expected 'Agent_#', got: {agent_name}"
)
# Step 2: Validate agent exists
if not agent_id:
raise ValidationError(f"Agent {agent_name} not found")
# Step 3: Validate permissions
await self.validate_deletion_permissions(
agent_id=agent_id,
session_id=session_id,
security_context=security_context,
force=force,
)
# Step 4: Validate agent state (if available)
if agent_state:
await self.validate_agent_state_for_deletion(
agent_state=agent_state, force=force
)
# All validations passed
self._validation_stats["successful_validations"] += 1
if self._audit_logger:
await self._audit_logger.log_event(
level=AuditLevel.INFO,
category=AuditCategory.VALIDATION,
operation="agent_deletion_validation_success",
resource_type="agent",
resource_id=str(agent_id),
success=True,
metadata={
"agent_name": agent_name,
"session_id": session_id,
"force": force,
"validation_checks": 4,
},
)
return True
except (ValidationError, SecurityError) as e:
self._validation_stats["failed_validations"] += 1
if self._audit_logger:
await self._audit_logger.log_event(
level=AuditLevel.WARNING,
category=AuditCategory.VALIDATION,
operation="agent_deletion_validation_failed",
resource_type="agent_deletion_request",
resource_id=agent_name,
success=False,
error_message=str(e),
metadata={
"session_id": session_id,
"force": force,
"error_type": type(e).__name__,
},
)
raise
def get_validation_stats(self) -> dict:
"""Get validation statistics."""
return {
**self._validation_stats,
"success_rate": (
self._validation_stats["successful_validations"]
/ self._validation_stats["total_validations"]
* 100
if self._validation_stats["total_validations"] > 0
else 0.0
),
}
def validate_agent_deletion_request(func: Callable) -> Callable:
"""
Decorator for agent deletion request validation.
Validates agent name format and basic request structure before
allowing deletion operation to proceed.
Args:
func: Function to wrap with validation
Returns:
Wrapped function with validation
"""
@wraps(func)
async def wrapper(*args: Any, **kwargs: Any) -> Any:
# Extract parameters
agent_name = kwargs.get("agent_name", args[1] if len(args) > 1 else None)
# Validate agent name format
if not agent_name or not isinstance(agent_name, str):
raise ValidationError("Agent name must be a non-empty string")
if not AGENT_NAME_PATTERN.match(agent_name):
raise ValidationError(
f"Invalid agent name format. Expected 'Agent_#', got: {agent_name}"
)
# Additional validation will be done in the tool implementation
# after we can verify the agent exists and check its state
return await func(*args, **kwargs)
return wrapper
@require(lambda agent_name: agent_name and isinstance(agent_name, str))
@ensure(lambda result: isinstance(result, bool))
def is_valid_agent_name(agent_name: str) -> bool:
"""
Check if agent name is valid format.
Args:
agent_name: Agent name to validate
Returns:
bool: True if valid format
"""
return bool(AGENT_NAME_PATTERN.match(agent_name))
# Export validation utilities
__all__ = [
"AgentDeletionValidator",
"validate_agent_deletion_request",
"is_valid_agent_name",
"AGENT_NAME_PATTERN",
]
