mirror of
https://github.com/crewAIInc/crewAI.git
synced 2026-05-02 07:42:40 +00:00
adding fingerprints (#2332)
* adding fingerprints * fixed * fix * Fix Pydantic v2 compatibility in SecurityConfig and Fingerprint classes (#2335) * Fix Pydantic v2 compatibility in SecurityConfig and Fingerprint classes Co-Authored-By: Joe Moura <joao@crewai.com> * Fix type-checker errors in fingerprint properties Co-Authored-By: Joe Moura <joao@crewai.com> * Enhance security validation in Fingerprint and SecurityConfig classes Co-Authored-By: Joe Moura <joao@crewai.com> --------- Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Co-authored-by: Joe Moura <joao@crewai.com> * incorporate small improvements / changes * Expect different * Remove redundant null check in Crew.fingerprint property (#2342) * Remove redundant null check in Crew.fingerprint property and add security module Co-Authored-By: Joe Moura <joao@crewai.com> * Enhance security module with type hints, improved UUID namespace, metadata validation, and versioning Co-Authored-By: Joe Moura <joao@crewai.com> --------- Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Co-authored-by: Joe Moura <joao@crewai.com> Co-authored-by: João Moura <joaomdmoura@gmail.com> --------- Co-authored-by: devin-ai-integration[bot] <158243242+devin-ai-integration[bot]@users.noreply.github.com> Co-authored-by: Joe Moura <joao@crewai.com> Co-authored-by: Brandon Hancock <brandon@brandonhancock.io>
This commit is contained in:
13
src/crewai/security/__init__.py
Normal file
13
src/crewai/security/__init__.py
Normal file
@@ -0,0 +1,13 @@
|
||||
"""
|
||||
CrewAI security module.
|
||||
|
||||
This module provides security-related functionality for CrewAI, including:
|
||||
- Fingerprinting for component identity and tracking
|
||||
- Security configuration for controlling access and permissions
|
||||
- Future: authentication, scoping, and delegation mechanisms
|
||||
"""
|
||||
|
||||
from crewai.security.fingerprint import Fingerprint
|
||||
from crewai.security.security_config import SecurityConfig
|
||||
|
||||
__all__ = ["Fingerprint", "SecurityConfig"]
|
||||
170
src/crewai/security/fingerprint.py
Normal file
170
src/crewai/security/fingerprint.py
Normal file
@@ -0,0 +1,170 @@
|
||||
"""
|
||||
Fingerprint Module
|
||||
|
||||
This module provides functionality for generating and validating unique identifiers
|
||||
for CrewAI agents. These identifiers are used for tracking, auditing, and security.
|
||||
"""
|
||||
|
||||
import uuid
|
||||
from datetime import datetime
|
||||
from typing import Any, Dict, Optional
|
||||
|
||||
from pydantic import BaseModel, ConfigDict, Field, field_validator
|
||||
|
||||
|
||||
class Fingerprint(BaseModel):
|
||||
"""
|
||||
A class for generating and managing unique identifiers for agents.
|
||||
|
||||
Each agent has dual identifiers:
|
||||
- Human-readable ID: For debugging and reference (derived from role if not specified)
|
||||
- Fingerprint UUID: Unique runtime identifier for tracking and auditing
|
||||
|
||||
Attributes:
|
||||
uuid_str (str): String representation of the UUID for this fingerprint, auto-generated
|
||||
created_at (datetime): When this fingerprint was created, auto-generated
|
||||
metadata (Dict[str, Any]): Additional metadata associated with this fingerprint
|
||||
"""
|
||||
|
||||
uuid_str: str = Field(default_factory=lambda: str(uuid.uuid4()), description="String representation of the UUID")
|
||||
created_at: datetime = Field(default_factory=datetime.now, description="When this fingerprint was created")
|
||||
metadata: Dict[str, Any] = Field(default_factory=dict, description="Additional metadata for this fingerprint")
|
||||
|
||||
model_config = ConfigDict(arbitrary_types_allowed=True)
|
||||
|
||||
@field_validator('metadata')
|
||||
@classmethod
|
||||
def validate_metadata(cls, v):
|
||||
"""Validate that metadata is a dictionary with string keys and valid values."""
|
||||
if not isinstance(v, dict):
|
||||
raise ValueError("Metadata must be a dictionary")
|
||||
|
||||
# Validate that all keys are strings
|
||||
for key, value in v.items():
|
||||
if not isinstance(key, str):
|
||||
raise ValueError(f"Metadata keys must be strings, got {type(key)}")
|
||||
|
||||
# Validate nested dictionaries (prevent deeply nested structures)
|
||||
if isinstance(value, dict):
|
||||
# Check for nested dictionaries (limit depth to 1)
|
||||
for nested_key, nested_value in value.items():
|
||||
if not isinstance(nested_key, str):
|
||||
raise ValueError(f"Nested metadata keys must be strings, got {type(nested_key)}")
|
||||
if isinstance(nested_value, dict):
|
||||
raise ValueError("Metadata can only be nested one level deep")
|
||||
|
||||
# Check for maximum metadata size (prevent DoS)
|
||||
if len(str(v)) > 10000: # Limit metadata size to 10KB
|
||||
raise ValueError("Metadata size exceeds maximum allowed (10KB)")
|
||||
|
||||
return v
|
||||
|
||||
def __init__(self, **data):
|
||||
"""Initialize a Fingerprint with auto-generated uuid_str and created_at."""
|
||||
# Remove uuid_str and created_at from data to ensure they're auto-generated
|
||||
if 'uuid_str' in data:
|
||||
data.pop('uuid_str')
|
||||
if 'created_at' in data:
|
||||
data.pop('created_at')
|
||||
|
||||
# Call the parent constructor with the modified data
|
||||
super().__init__(**data)
|
||||
|
||||
@property
|
||||
def uuid(self) -> uuid.UUID:
|
||||
"""Get the UUID object for this fingerprint."""
|
||||
return uuid.UUID(self.uuid_str)
|
||||
|
||||
@classmethod
|
||||
def _generate_uuid(cls, seed: str) -> str:
|
||||
"""
|
||||
Generate a deterministic UUID based on a seed string.
|
||||
|
||||
Args:
|
||||
seed (str): The seed string to use for UUID generation
|
||||
|
||||
Returns:
|
||||
str: A string representation of the UUID consistently generated from the seed
|
||||
"""
|
||||
if not isinstance(seed, str):
|
||||
raise ValueError("Seed must be a string")
|
||||
|
||||
if not seed.strip():
|
||||
raise ValueError("Seed cannot be empty or whitespace")
|
||||
|
||||
# Create a deterministic UUID using v5 (SHA-1)
|
||||
# Custom namespace for CrewAI to enhance security
|
||||
|
||||
# Using a unique namespace specific to CrewAI to reduce collision risks
|
||||
CREW_AI_NAMESPACE = uuid.UUID('f47ac10b-58cc-4372-a567-0e02b2c3d479')
|
||||
return str(uuid.uuid5(CREW_AI_NAMESPACE, seed))
|
||||
|
||||
@classmethod
|
||||
def generate(cls, seed: Optional[str] = None, metadata: Optional[Dict[str, Any]] = None) -> 'Fingerprint':
|
||||
"""
|
||||
Static factory method to create a new Fingerprint.
|
||||
|
||||
Args:
|
||||
seed (Optional[str]): A string to use as seed for the UUID generation.
|
||||
If None, a random UUID is generated.
|
||||
metadata (Optional[Dict[str, Any]]): Additional metadata to store with the fingerprint.
|
||||
|
||||
Returns:
|
||||
Fingerprint: A new Fingerprint instance
|
||||
"""
|
||||
fingerprint = cls(metadata=metadata or {})
|
||||
if seed:
|
||||
# For seed-based generation, we need to manually set the uuid_str after creation
|
||||
object.__setattr__(fingerprint, 'uuid_str', cls._generate_uuid(seed))
|
||||
return fingerprint
|
||||
|
||||
def __str__(self) -> str:
|
||||
"""String representation of the fingerprint (the UUID)."""
|
||||
return self.uuid_str
|
||||
|
||||
def __eq__(self, other) -> bool:
|
||||
"""Compare fingerprints by their UUID."""
|
||||
if isinstance(other, Fingerprint):
|
||||
return self.uuid_str == other.uuid_str
|
||||
return False
|
||||
|
||||
def __hash__(self) -> int:
|
||||
"""Hash of the fingerprint (based on UUID)."""
|
||||
return hash(self.uuid_str)
|
||||
|
||||
def to_dict(self) -> Dict[str, Any]:
|
||||
"""
|
||||
Convert the fingerprint to a dictionary representation.
|
||||
|
||||
Returns:
|
||||
Dict[str, Any]: Dictionary representation of the fingerprint
|
||||
"""
|
||||
return {
|
||||
"uuid_str": self.uuid_str,
|
||||
"created_at": self.created_at.isoformat(),
|
||||
"metadata": self.metadata
|
||||
}
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, data: Dict[str, Any]) -> 'Fingerprint':
|
||||
"""
|
||||
Create a Fingerprint from a dictionary representation.
|
||||
|
||||
Args:
|
||||
data (Dict[str, Any]): Dictionary representation of a fingerprint
|
||||
|
||||
Returns:
|
||||
Fingerprint: A new Fingerprint instance
|
||||
"""
|
||||
if not data:
|
||||
return cls()
|
||||
|
||||
fingerprint = cls(metadata=data.get("metadata", {}))
|
||||
|
||||
# For consistency with existing stored fingerprints, we need to manually set these
|
||||
if "uuid_str" in data:
|
||||
object.__setattr__(fingerprint, 'uuid_str', data["uuid_str"])
|
||||
if "created_at" in data and isinstance(data["created_at"], str):
|
||||
object.__setattr__(fingerprint, 'created_at', datetime.fromisoformat(data["created_at"]))
|
||||
|
||||
return fingerprint
|
||||
116
src/crewai/security/security_config.py
Normal file
116
src/crewai/security/security_config.py
Normal file
@@ -0,0 +1,116 @@
|
||||
"""
|
||||
Security Configuration Module
|
||||
|
||||
This module provides configuration for CrewAI security features, including:
|
||||
- Authentication settings
|
||||
- Scoping rules
|
||||
- Fingerprinting
|
||||
|
||||
The SecurityConfig class is the primary interface for managing security settings
|
||||
in CrewAI applications.
|
||||
"""
|
||||
|
||||
from typing import Any, Dict, Optional
|
||||
|
||||
from pydantic import BaseModel, ConfigDict, Field, model_validator
|
||||
|
||||
from crewai.security.fingerprint import Fingerprint
|
||||
|
||||
|
||||
class SecurityConfig(BaseModel):
|
||||
"""
|
||||
Configuration for CrewAI security features.
|
||||
|
||||
This class manages security settings for CrewAI agents, including:
|
||||
- Authentication credentials *TODO*
|
||||
- Identity information (agent fingerprints)
|
||||
- Scoping rules *TODO*
|
||||
- Impersonation/delegation tokens *TODO*
|
||||
|
||||
Attributes:
|
||||
version (str): Version of the security configuration
|
||||
fingerprint (Fingerprint): The unique fingerprint automatically generated for the component
|
||||
"""
|
||||
|
||||
model_config = ConfigDict(
|
||||
arbitrary_types_allowed=True
|
||||
# Note: Cannot use frozen=True as existing tests modify the fingerprint property
|
||||
)
|
||||
|
||||
version: str = Field(
|
||||
default="1.0.0",
|
||||
description="Version of the security configuration"
|
||||
)
|
||||
|
||||
fingerprint: Fingerprint = Field(
|
||||
default_factory=Fingerprint,
|
||||
description="Unique identifier for the component"
|
||||
)
|
||||
|
||||
def is_compatible(self, min_version: str) -> bool:
|
||||
"""
|
||||
Check if this security configuration is compatible with the minimum required version.
|
||||
|
||||
Args:
|
||||
min_version (str): Minimum required version in semver format (e.g., "1.0.0")
|
||||
|
||||
Returns:
|
||||
bool: True if this configuration is compatible, False otherwise
|
||||
"""
|
||||
# Simple version comparison (can be enhanced with packaging.version if needed)
|
||||
current = [int(x) for x in self.version.split(".")]
|
||||
minimum = [int(x) for x in min_version.split(".")]
|
||||
|
||||
# Compare major, minor, patch versions
|
||||
for c, m in zip(current, minimum):
|
||||
if c > m:
|
||||
return True
|
||||
if c < m:
|
||||
return False
|
||||
return True
|
||||
|
||||
@model_validator(mode='before')
|
||||
@classmethod
|
||||
def validate_fingerprint(cls, values):
|
||||
"""Ensure fingerprint is properly initialized."""
|
||||
if isinstance(values, dict):
|
||||
# Handle case where fingerprint is not provided or is None
|
||||
if 'fingerprint' not in values or values['fingerprint'] is None:
|
||||
values['fingerprint'] = Fingerprint()
|
||||
# Handle case where fingerprint is a string (seed)
|
||||
elif isinstance(values['fingerprint'], str):
|
||||
if not values['fingerprint'].strip():
|
||||
raise ValueError("Fingerprint seed cannot be empty")
|
||||
values['fingerprint'] = Fingerprint.generate(seed=values['fingerprint'])
|
||||
return values
|
||||
|
||||
def to_dict(self) -> Dict[str, Any]:
|
||||
"""
|
||||
Convert the security config to a dictionary.
|
||||
|
||||
Returns:
|
||||
Dict[str, Any]: Dictionary representation of the security config
|
||||
"""
|
||||
result = {
|
||||
"fingerprint": self.fingerprint.to_dict()
|
||||
}
|
||||
return result
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, data: Dict[str, Any]) -> 'SecurityConfig':
|
||||
"""
|
||||
Create a SecurityConfig from a dictionary.
|
||||
|
||||
Args:
|
||||
data (Dict[str, Any]): Dictionary representation of a security config
|
||||
|
||||
Returns:
|
||||
SecurityConfig: A new SecurityConfig instance
|
||||
"""
|
||||
# Make a copy to avoid modifying the original
|
||||
data_copy = data.copy()
|
||||
|
||||
fingerprint_data = data_copy.pop("fingerprint", None)
|
||||
fingerprint = Fingerprint.from_dict(fingerprint_data) if fingerprint_data else Fingerprint()
|
||||
|
||||
return cls(fingerprint=fingerprint)
|
||||
Reference in New Issue
Block a user