Adding Defender code

src/llmguardian/defenders/__init__.py

@@ -0,0 +1,17 @@
+"""
+defenders/__init__.py - Security defenders initialization
+"""
+
+from .input_sanitizer import InputSanitizer
+from .output_validator import OutputValidator
+from .token_validator import TokenValidator
+from .content_filter import ContentFilter
+from .context_validator import ContextValidator
+
+__all__ = [
+    'InputSanitizer',
+    'OutputValidator',
+    'TokenValidator',
+    'ContentFilter',
+    'ContextValidator',
+]

src/llmguardian/defenders/content_filter.py

@@ -0,0 +1,167 @@
+"""
+defenders/content_filter.py - Content filtering and moderation
+"""
+
+import re
+from typing import Dict, List, Optional, Any, Set
+from dataclasses import dataclass
+from enum import Enum
+from ..core.logger import SecurityLogger
+from ..core.exceptions import ValidationError
+
+class ContentCategory(Enum):
+    MALICIOUS = "malicious"
+    SENSITIVE = "sensitive"
+    HARMFUL = "harmful"
+    INAPPROPRIATE = "inappropriate"
+    POTENTIAL_EXPLOIT = "potential_exploit"
+
+@dataclass
+class FilterRule:
+    pattern: str
+    category: ContentCategory
+    severity: int  # 1-10
+    description: str
+    action: str  # "block" or "sanitize"
+    replacement: str = "[FILTERED]"
+
+@dataclass
+class FilterResult:
+    is_allowed: bool
+    filtered_content: str
+    matched_rules: List[str]
+    risk_score: int
+    categories: Set[ContentCategory]
+    details: Dict[str, Any]
+
+class ContentFilter:
+    def __init__(self, security_logger: Optional[SecurityLogger] = None):
+        self.security_logger = security_logger
+        self.rules = self._initialize_rules()
+        self.compiled_rules = {
+            name: re.compile(rule.pattern, re.IGNORECASE | re.MULTILINE)
+            for name, rule in self.rules.items()
+        }
+
+    def _initialize_rules(self) -> Dict[str, FilterRule]:
+        return {
+            "code_execution": FilterRule(
+                pattern=r"(?:exec|eval|system|subprocess|os\.)",
+                category=ContentCategory.MALICIOUS,
+                severity=9,
+                description="Code execution attempt",
+                action="block"
+            ),
+            "sql_commands": FilterRule(
+                pattern=r"(?:SELECT|INSERT|UPDATE|DELETE|DROP|UNION)\s+(?:FROM|INTO|TABLE)",
+                category=ContentCategory.MALICIOUS,
+                severity=8,
+                description="SQL command",
+                action="block"
+            ),
+            "file_operations": FilterRule(
+                pattern=r"(?:read|write|open|delete|remove)\s*\(['\"].*?['\"]",
+                category=ContentCategory.POTENTIAL_EXPLOIT,
+                severity=7,
+                description="File operation",
+                action="block"
+            ),
+            "pii_data": FilterRule(
+                pattern=r"\b\d{3}-\d{2}-\d{4}\b|\b\d{16}\b",
+                category=ContentCategory.SENSITIVE,
+                severity=8,
+                description="PII data",
+                action="sanitize",
+                replacement="[REDACTED]"
+            ),
+            "harmful_content": FilterRule(
+                pattern=r"(?:hack|exploit|bypass|vulnerability)\s+(?:system|security|protection)",
+                category=ContentCategory.HARMFUL,
+                severity=7,
+                description="Potentially harmful content",
+                action="block"
+            ),
+            "inappropriate_content": FilterRule(
+                pattern=r"(?:explicit|offensive|inappropriate).*content",
+                category=ContentCategory.INAPPROPRIATE,
+                severity=6,
+                description="Inappropriate content",
+                action="sanitize"
+            ),
+        }
+
+    def filter_content(self, content: str, context: Optional[Dict[str, Any]] = None) -> FilterResult:
+        try:
+            matched_rules = []
+            categories = set()
+            risk_score = 0
+            filtered = content
+            is_allowed = True
+
+            for name, rule in self.rules.items():
+                pattern = self.compiled_rules[name]
+                matches = pattern.findall(filtered)
+
+                if matches:
+                    matched_rules.append(name)
+                    categories.add(rule.category)
+                    risk_score = max(risk_score, rule.severity)
+
+                    if rule.action == "block":
+                        is_allowed = False
+                    elif rule.action == "sanitize":
+                        filtered = pattern.sub(rule.replacement, filtered)
+
+            result = FilterResult(
+                is_allowed=is_allowed,
+                filtered_content=filtered if is_allowed else "[CONTENT BLOCKED]",
+                matched_rules=matched_rules,
+                risk_score=risk_score,
+                categories=categories,
+                details={
+                    "original_length": len(content),
+                    "filtered_length": len(filtered),
+                    "rule_matches": len(matched_rules),
+                    "context": context or {}
+                }
+            )
+
+            if matched_rules and self.security_logger:
+                self.security_logger.log_security_event(
+                    "content_filtered",
+                    matched_rules=matched_rules,
+                    categories=[c.value for c in categories],
+                    risk_score=risk_score,
+                    is_allowed=is_allowed
+                )
+
+            return result
+
+        except Exception as e:
+            if self.security_logger:
+                self.security_logger.log_security_event(
+                    "filter_error",
+                    error=str(e),
+                    content_length=len(content)
+                )
+            raise ValidationError(f"Content filtering failed: {str(e)}")
+
+    def add_rule(self, name: str, rule: FilterRule) -> None:
+        self.rules[name] = rule
+        self.compiled_rules[name] = re.compile(rule.pattern, re.IGNORECASE | re.MULTILINE)
+
+    def remove_rule(self, name: str) -> None:
+        self.rules.pop(name, None)
+        self.compiled_rules.pop(name, None)
+
+    def get_rules(self) -> Dict[str, Dict[str, Any]]:
+        return {
+            name: {
+                "pattern": rule.pattern,
+                "category": rule.category.value,
+                "severity": rule.severity,
+                "description": rule.description,
+                "action": rule.action
+            }
+            for name, rule in self.rules.items()
+        }

src/llmguardian/defenders/context_validator.py

@@ -0,0 +1,123 @@
+"""
+defenders/context_validator.py - Context validation for LLM interactions
+"""
+
+from typing import Dict, Optional, List, Any
+from dataclasses import dataclass
+from datetime import datetime
+import hashlib
+from ..core.logger import SecurityLogger
+from ..core.exceptions import ValidationError
+
+@dataclass
+class ContextRule:
+   max_age: int  # seconds
+   required_fields: List[str] 
+   forbidden_fields: List[str]
+   max_depth: int
+   checksum_fields: List[str]
+
+@dataclass
+class ValidationResult:
+   is_valid: bool
+   errors: List[str]
+   modified_context: Dict[str, Any]
+   metadata: Dict[str, Any]
+
+class ContextValidator:
+   def __init__(self, security_logger: Optional[SecurityLogger] = None):
+       self.security_logger = security_logger
+       self.rule = ContextRule(
+           max_age=3600,
+           required_fields=["user_id", "session_id", "timestamp"],
+           forbidden_fields=["password", "secret", "token"],
+           max_depth=5,
+           checksum_fields=["user_id", "session_id"]
+       )
+
+   def validate_context(self, context: Dict[str, Any], previous_context: Optional[Dict[str, Any]] = None) -> ValidationResult:
+       try:
+           errors = []
+           modified = context.copy()
+
+           # Check required fields
+           missing = [f for f in self.rule.required_fields if f not in context]
+           if missing:
+               errors.append(f"Missing required fields: {missing}")
+
+           # Check forbidden fields
+           forbidden = [f for f in self.rule.forbidden_fields if f in context]
+           if forbidden:
+               errors.append(f"Forbidden fields present: {forbidden}")
+               for field in forbidden:
+                   modified.pop(field, None)
+
+           # Validate timestamp
+           if "timestamp" in context:
+               age = (datetime.utcnow() - datetime.fromisoformat(str(context["timestamp"]))).seconds
+               if age > self.rule.max_age:
+                   errors.append(f"Context too old: {age} seconds")
+
+           # Check context depth
+           if not self._check_depth(context, 0):
+               errors.append(f"Context exceeds max depth of {self.rule.max_depth}")
+
+           # Verify checksums if previous context exists
+           if previous_context:
+               if not self._verify_checksums(context, previous_context):
+                   errors.append("Context checksum mismatch")
+
+           # Build metadata
+           metadata = {
+               "validation_time": datetime.utcnow().isoformat(),
+               "original_size": len(str(context)),
+               "modified_size": len(str(modified)),
+               "changes": len(errors)
+           }
+
+           result = ValidationResult(
+               is_valid=len(errors) == 0,
+               errors=errors,
+               modified_context=modified,
+               metadata=metadata
+           )
+
+           if errors and self.security_logger:
+               self.security_logger.log_security_event(
+                   "context_validation_failure",
+                   errors=errors,
+                   context_id=context.get("context_id")
+               )
+
+           return result
+
+       except Exception as e:
+           if self.security_logger:
+               self.security_logger.log_security_event(
+                   "context_validation_error",
+                   error=str(e)
+               )
+           raise ValidationError(f"Context validation failed: {str(e)}")
+
+   def _check_depth(self, obj: Any, depth: int) -> bool:
+       if depth > self.rule.max_depth:
+           return False
+       if isinstance(obj, dict):
+           return all(self._check_depth(v, depth + 1) for v in obj.values())
+       if isinstance(obj, list):
+           return all(self._check_depth(v, depth + 1) for v in obj)
+       return True
+
+   def _verify_checksums(self, current: Dict[str, Any], previous: Dict[str, Any]) -> bool:
+       for field in self.rule.checksum_fields:
+           if field in current and field in previous:
+               current_hash = hashlib.sha256(str(current[field]).encode()).hexdigest()
+               previous_hash = hashlib.sha256(str(previous[field]).encode()).hexdigest()
+               if current_hash != previous_hash:
+                   return False
+       return True
+
+   def update_rule(self, updates: Dict[str, Any]) -> None:
+       for key, value in updates.items():
+           if hasattr(self.rule, key):
+               setattr(self.rule, key, value)

src/llmguardian/defenders/input_sanitizer.py

@@ -0,0 +1,141 @@
+"""
+defenders/input_sanitizer.py - Input sanitization for LLM inputs
+"""
+
+import re
+from typing import Dict, Any, List, Optional
+from dataclasses import dataclass
+from ..core.logger import SecurityLogger
+from ..core.exceptions import ValidationError
+
+@dataclass
+class SanitizationRule:
+    pattern: str
+    replacement: str
+    description: str
+    enabled: bool = True
+
+@dataclass
+class SanitizationResult:
+    original: str
+    sanitized: str
+    applied_rules: List[str]
+    is_modified: bool
+    risk_level: str
+
+class InputSanitizer:
+    def __init__(self, security_logger: Optional[SecurityLogger] = None):
+        self.security_logger = security_logger
+        self.rules = self._initialize_rules()
+        self.compiled_rules = {
+            name: re.compile(rule.pattern, re.IGNORECASE | re.MULTILINE)
+            for name, rule in self.rules.items()
+            if rule.enabled
+        }
+
+    def _initialize_rules(self) -> Dict[str, SanitizationRule]:
+        return {
+            "system_instructions": SanitizationRule(
+                pattern=r"system:\s*|instruction:\s*",
+                replacement=" ",
+                description="Remove system instruction markers"
+            ),
+            "code_injection": SanitizationRule(
+                pattern=r"<script.*?>.*?</script>",
+                replacement="",
+                description="Remove script tags"
+            ),
+            "delimiter_injection": SanitizationRule(
+                pattern=r"[<\[{](?:system|prompt|instruction)[>\]}]",
+                replacement="",
+                description="Remove delimiter-based injections"
+            ),
+            "command_injection": SanitizationRule(
+                pattern=r"(?:exec|eval|system)\s*\(",
+                replacement="",
+                description="Remove command execution attempts"
+            ),
+            "encoding_patterns": SanitizationRule(
+                pattern=r"(?:base64|hex|rot13)\s*\(",
+                replacement="",
+                description="Remove encoding attempts"
+            ),
+        }
+
+    def sanitize(self, input_text: str, context: Optional[Dict[str, Any]] = None) -> SanitizationResult:
+        original = input_text
+        applied_rules = []
+        is_modified = False
+
+        try:
+            sanitized = input_text
+            for name, rule in self.rules.items():
+                if not rule.enabled:
+                    continue
+
+                pattern = self.compiled_rules.get(name)
+                if not pattern:
+                    continue
+
+                new_text = pattern.sub(rule.replacement, sanitized)
+                if new_text != sanitized:
+                    applied_rules.append(name)
+                    is_modified = True
+                    sanitized = new_text
+
+            risk_level = self._assess_risk(applied_rules)
+
+            if is_modified and self.security_logger:
+                self.security_logger.log_security_event(
+                    "input_sanitization",
+                    original_length=len(original),
+                    sanitized_length=len(sanitized),
+                    applied_rules=applied_rules,
+                    risk_level=risk_level
+                )
+
+            return SanitizationResult(
+                original=original,
+                sanitized=sanitized,
+                applied_rules=applied_rules,
+                is_modified=is_modified,
+                risk_level=risk_level
+            )
+
+        except Exception as e:
+            if self.security_logger:
+                self.security_logger.log_security_event(
+                    "sanitization_error",
+                    error=str(e),
+                    input_length=len(input_text)
+                )
+            raise ValidationError(f"Sanitization failed: {str(e)}")
+
+    def _assess_risk(self, applied_rules: List[str]) -> str:
+        if not applied_rules:
+            return "low"
+        if len(applied_rules) >= 3:
+            return "high"
+        if "command_injection" in applied_rules or "code_injection" in applied_rules:
+            return "high"
+        return "medium"
+
+    def add_rule(self, name: str, rule: SanitizationRule) -> None:
+        self.rules[name] = rule
+        if rule.enabled:
+            self.compiled_rules[name] = re.compile(rule.pattern, re.IGNORECASE | re.MULTILINE)
+
+    def remove_rule(self, name: str) -> None:
+        self.rules.pop(name, None)
+        self.compiled_rules.pop(name, None)
+
+    def get_rules(self) -> Dict[str, Dict[str, Any]]:
+        return {
+            name: {
+                "pattern": rule.pattern,
+                "replacement": rule.replacement,
+                "description": rule.description,
+                "enabled": rule.enabled
+            }
+            for name, rule in self.rules.items()
+        }

src/llmguardian/defenders/output_validator.py

@@ -0,0 +1,173 @@
+"""
+defenders/output_validator.py - Output validation and sanitization
+"""
+
+import re
+from typing import Dict, List, Optional, Set, Any
+from dataclasses import dataclass
+from ..core.logger import SecurityLogger
+from ..core.exceptions import ValidationError
+
+@dataclass
+class ValidationRule:
+    pattern: str
+    description: str
+    severity: int  # 1-10
+    block: bool = True
+    sanitize: bool = True
+    replacement: str = ""
+
+@dataclass
+class ValidationResult:
+    is_valid: bool
+    violations: List[str]
+    sanitized_output: Optional[str]
+    risk_score: int
+    details: Dict[str, Any]
+
+class OutputValidator:
+    def __init__(self, security_logger: Optional[SecurityLogger] = None):
+        self.security_logger = security_logger
+        self.rules = self._initialize_rules()
+        self.compiled_rules = {
+            name: re.compile(rule.pattern, re.IGNORECASE | re.MULTILINE)
+            for name, rule in self.rules.items()
+        }
+        self.sensitive_patterns = self._initialize_sensitive_patterns()
+
+    def _initialize_rules(self) -> Dict[str, ValidationRule]:
+        return {
+            "sql_injection": ValidationRule(
+                pattern=r"(?:SELECT|INSERT|UPDATE|DELETE)\s+(?:FROM|INTO)\s+\w+",
+                description="SQL query in output",
+                severity=9,
+                block=True
+            ),
+            "code_injection": ValidationRule(
+                pattern=r"<script.*?>.*?</script>",
+                description="JavaScript code in output",
+                severity=8,
+                block=True
+            ),
+            "system_info": ValidationRule(
+                pattern=r"(?:system|config|env|secret)(?:_|\s+)?(?:key|token|password)",
+                description="System information leak",
+                severity=9,
+                block=True
+            ),
+            "personal_data": ValidationRule(
+                pattern=r"\b\d{3}-\d{2}-\d{4}\b|\b\d{16}\b",
+                description="Personal data (SSN/CC)",
+                severity=10,
+                block=True
+            ),
+            "file_paths": ValidationRule(
+                pattern=r"(?:/[\w./]+)|(?:C:\\[\w\\]+)",
+                description="File system paths",
+                severity=7,
+                block=True
+            ),
+            "html_content": ValidationRule(
+                pattern=r"<(?!br|p|b|i|em|strong)[^>]+>",
+                description="HTML content",
+                severity=6,
+                sanitize=True,
+                replacement=""
+            ),
+        }
+
+    def _initialize_sensitive_patterns(self) -> Set[str]:
+        return {
+            r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b",  # Email
+            r"\b\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}\b",  # IP address
+            r"(?i)api[_-]?key",  # API keys
+            r"(?i)password|passwd|pwd",  # Passwords
+            r"(?i)token|secret|credential",  # Credentials
+            r"\b[A-Z0-9]{20,}\b",  # Long alphanumeric strings
+        }
+
+    def validate(self, output: str, context: Optional[Dict[str, Any]] = None) -> ValidationResult:
+        try:
+            violations = []
+            risk_score = 0
+            sanitized = output
+            is_valid = True
+
+            # Check against validation rules
+            for name, rule in self.rules.items():
+                pattern = self.compiled_rules[name]
+                matches = pattern.findall(sanitized)
+                
+                if matches:
+                    violations.append(f"{name}: {rule.description}")
+                    risk_score = max(risk_score, rule.severity)
+                    
+                    if rule.block:
+                        is_valid = False
+                    
+                    if rule.sanitize:
+                        sanitized = pattern.sub(rule.replacement, sanitized)
+
+            # Check for sensitive data patterns
+            for pattern in self.sensitive_patterns:
+                matches = re.findall(pattern, sanitized)
+                if matches:
+                    violations.append(f"Sensitive data detected: {pattern}")
+                    risk_score = max(risk_score, 8)
+                    is_valid = False
+                    sanitized = re.sub(pattern, "[REDACTED]", sanitized)
+
+            result = ValidationResult(
+                is_valid=is_valid,
+                violations=violations,
+                sanitized_output=sanitized if violations else output,
+                risk_score=risk_score,
+                details={
+                    "original_length": len(output),
+                    "sanitized_length": len(sanitized),
+                    "violation_count": len(violations),
+                    "context": context or {}
+                }
+            )
+
+            if violations and self.security_logger:
+                self.security_logger.log_security_event(
+                    "output_validation",
+                    violations=violations,
+                    risk_score=risk_score,
+                    is_valid=is_valid
+                )
+
+            return result
+
+        except Exception as e:
+            if self.security_logger:
+                self.security_logger.log_security_event(
+                    "validation_error",
+                    error=str(e),
+                    output_length=len(output)
+                )
+            raise ValidationError(f"Output validation failed: {str(e)}")
+
+    def add_rule(self, name: str, rule: ValidationRule) -> None:
+        self.rules[name] = rule
+        self.compiled_rules[name] = re.compile(rule.pattern, re.IGNORECASE | re.MULTILINE)
+
+    def remove_rule(self, name: str) -> None:
+        self.rules.pop(name, None)
+        self.compiled_rules.pop(name, None)
+
+    def add_sensitive_pattern(self, pattern: str) -> None:
+        self.sensitive_patterns.add(pattern)
+
+    def get_rules(self) -> Dict[str, Dict[str, Any]]:
+        return {
+            name: {
+                "pattern": rule.pattern,
+                "description": rule.description,
+                "severity": rule.severity,
+                "block": rule.block,
+                "sanitize": rule.sanitize
+            }
+            for name, rule in self.rules.items()
+        }

src/llmguardian/defenders/test_context_validator.py

@@ -0,0 +1,98 @@
+"""
+tests/defenders/test_context_validator.py - Tests for context validation
+"""
+
+import pytest
+from datetime import datetime, timedelta
+from llmguardian.defenders.context_validator import ContextValidator, ValidationResult
+from llmguardian.core.exceptions import ValidationError
+
+@pytest.fixture
+def validator():
+    return ContextValidator()
+
+@pytest.fixture
+def valid_context():
+    return {
+        "user_id": "test_user",
+        "session_id": "test_session",
+        "timestamp": datetime.utcnow().isoformat(),
+        "request_id": "123",
+        "metadata": {
+            "source": "test",
+            "version": "1.0"
+        }
+    }
+
+def test_valid_context(validator, valid_context):
+    result = validator.validate_context(valid_context)
+    assert result.is_valid
+    assert not result.errors
+    assert result.modified_context == valid_context
+
+def test_missing_required_fields(validator):
+    context = {
+        "user_id": "test_user",
+        "timestamp": datetime.utcnow().isoformat()
+    }
+    result = validator.validate_context(context)
+    assert not result.is_valid
+    assert "Missing required fields" in result.errors[0]
+
+def test_forbidden_fields(validator, valid_context):
+    context = valid_context.copy()
+    context["password"] = "secret123"
+    result = validator.validate_context(context)
+    assert not result.is_valid
+    assert "Forbidden fields present" in result.errors[0]
+    assert "password" not in result.modified_context
+
+def test_context_age(validator, valid_context):
+    old_context = valid_context.copy()
+    old_context["timestamp"] = (
+        datetime.utcnow() - timedelta(hours=2)
+    ).isoformat()
+    result = validator.validate_context(old_context)
+    assert not result.is_valid
+    assert "Context too old" in result.errors[0]
+
+def test_context_depth(validator, valid_context):
+    deep_context = valid_context.copy()
+    current = deep_context
+    for i in range(10):
+        current["nested"] = {}
+        current = current["nested"]
+    result = validator.validate_context(deep_context)
+    assert not result.is_valid
+    assert "Context exceeds max depth" in result.errors[0]
+
+def test_checksum_verification(validator, valid_context):
+    previous_context = valid_context.copy()
+    modified_context = valid_context.copy()
+    modified_context["user_id"] = "different_user"
+    result = validator.validate_context(modified_context, previous_context)
+    assert not result.is_valid
+    assert "Context checksum mismatch" in result.errors[0]
+
+def test_update_rule(validator):
+    validator.update_rule({"max_age": 7200})
+    old_context = {
+        "user_id": "test_user",
+        "session_id": "test_session",
+        "timestamp": (
+            datetime.utcnow() - timedelta(hours=1.5)
+        ).isoformat()
+    }
+    result = validator.validate_context(old_context)
+    assert result.is_valid
+
+def test_exception_handling(validator):
+    with pytest.raises(ValidationError):
+        validator.validate_context({"timestamp": "invalid_date"})
+
+def test_metadata_generation(validator, valid_context):
+    result = validator.validate_context(valid_context)
+    assert "validation_time" in result.metadata
+    assert "original_size" in result.metadata
+    assert "modified_size" in result.metadata
+    assert "changes" in result.metadata

src/llmguardian/defenders/token_validator.py

@@ -0,0 +1,147 @@
+"""
+defenders/token_validator.py - Token and credential validation
+"""
+
+from typing import Dict, Optional, Any, List
+from dataclasses import dataclass
+import re
+import jwt
+from datetime import datetime, timedelta
+from ..core.logger import SecurityLogger
+from ..core.exceptions import TokenValidationError
+
+@dataclass
+class TokenRule:
+    pattern: str
+    description: str
+    min_length: int
+    max_length: int
+    required_chars: str
+    expiry_time: int  # in seconds
+
+@dataclass
+class TokenValidationResult:
+    is_valid: bool
+    errors: List[str]
+    metadata: Dict[str, Any]
+    expiry: Optional[datetime]
+
+class TokenValidator:
+    def __init__(self, security_logger: Optional[SecurityLogger] = None):
+        self.security_logger = security_logger
+        self.rules = self._initialize_rules()
+        self.secret_key = self._load_secret_key()
+
+    def _initialize_rules(self) -> Dict[str, TokenRule]:
+        return {
+            "jwt": TokenRule(
+                pattern=r"^[A-Za-z0-9-_=]+\.[A-Za-z0-9-_=]+\.[A-Za-z0-9-_.+/=]+$",
+                description="JWT token",
+                min_length=32,
+                max_length=4096,
+                required_chars=".-_",
+                expiry_time=3600
+            ),
+            "api_key": TokenRule(
+                pattern=r"^[A-Za-z0-9]{32,64}$",
+                description="API key",
+                min_length=32,
+                max_length=64,
+                required_chars="",
+                expiry_time=86400
+            ),
+            "session_token": TokenRule(
+                pattern=r"^[A-Fa-f0-9]{64}$",
+                description="Session token",
+                min_length=64,
+                max_length=64,
+                required_chars="",
+                expiry_time=7200
+            )
+        }
+
+    def _load_secret_key(self) -> bytes:
+        # Implementation would load from secure storage
+        return b"your-256-bit-secret"
+
+    def validate_token(self, token: str, token_type: str) -> TokenValidationResult:
+        try:
+            if token_type not in self.rules:
+                raise TokenValidationError(f"Unknown token type: {token_type}")
+
+            rule = self.rules[token_type]
+            errors = []
+            metadata = {}
+
+            # Length validation
+            if len(token) < rule.min_length or len(token) > rule.max_length:
+                errors.append(f"Token length must be between {rule.min_length} and {rule.max_length}")
+
+            # Pattern validation
+            if not re.match(rule.pattern, token):
+                errors.append("Token format is invalid")
+
+            # Required characters
+            if rule.required_chars:
+                missing_chars = set(rule.required_chars) - set(token)
+                if missing_chars:
+                    errors.append(f"Token missing required characters: {missing_chars}")
+
+            # JWT-specific validation
+            if token_type == "jwt":
+                try:
+                    payload = jwt.decode(token, self.secret_key, algorithms=["HS256"])
+                    metadata = payload
+                    exp = datetime.fromtimestamp(payload.get("exp", 0))
+                    if exp < datetime.utcnow():
+                        errors.append("Token has expired")
+                except jwt.InvalidTokenError as e:
+                    errors.append(f"Invalid JWT: {str(e)}")
+
+            is_valid = len(errors) == 0
+            expiry = datetime.utcnow() + timedelta(seconds=rule.expiry_time)
+
+            if not is_valid and self.security_logger:
+                self.security_logger.log_security_event(
+                    "token_validation_failure",
+                    token_type=token_type,
+                    errors=errors
+                )
+
+            return TokenValidationResult(
+                is_valid=is_valid,
+                errors=errors,
+                metadata=metadata,
+                expiry=expiry if is_valid else None
+            )
+
+        except Exception as e:
+            if self.security_logger:
+                self.security_logger.log_security_event(
+                    "token_validation_error",
+                    error=str(e)
+                )
+            raise TokenValidationError(f"Validation failed: {str(e)}")
+
+    def create_token(self, token_type: str, payload: Dict[str, Any]) -> str:
+        if token_type not in self.rules:
+            raise TokenValidationError(f"Unknown token type: {token_type}")
+
+        try:
+            if token_type == "jwt":
+                expiry = datetime.utcnow() + timedelta(
+                    seconds=self.rules[token_type].expiry_time
+                )
+                payload["exp"] = expiry.timestamp()
+                return jwt.encode(payload, self.secret_key, algorithm="HS256")
+
+            # Add other token type creation logic here
+            raise TokenValidationError(f"Token creation not implemented for {token_type}")
+
+        except Exception as e:
+            if self.security_logger:
+                self.security_logger.log_security_event(
+                    "token_creation_error",
+                    error=str(e)
+                )
+            raise TokenValidationError(f"Token creation failed: {str(e)}")