User Isolation

What is User Isolation?

User Isolation is a fundamental security principle built into every layer of the Fiberwise platform. It ensures that users can only access their own data, preventing unauthorized cross-user data access while maintaining system performance and usability.

Isolation Architecture

Data Isolation Implementation

Database-Level Isolation

-- All data tables include user_id for automatic filtering
CREATE TABLE user_documents (
    id SERIAL PRIMARY KEY,
    user_id UUID NOT NULL REFERENCES users(id),
    filename VARCHAR(255) NOT NULL,
    content TEXT,
    created_at TIMESTAMP DEFAULT NOW(),
    updated_at TIMESTAMP DEFAULT NOW()
);

-- Automatic user context in all queries
-- This query is automatically modified by the platform:
-- SELECT * FROM user_documents WHERE id = ?
-- Becomes:
-- SELECT * FROM user_documents WHERE id = ? AND user_id = ?

ORM-Level Protection

# models.py - User isolation built into models
from fiberwise_sdk import UserIsolatedModel

class Document(UserIsolatedModel):
    """Documents are automatically isolated by user"""
    filename = models.CharField(max_length=255)
    content = models.TextField()
    created_at = models.DateTimeField(auto_now_add=True)
    
    # user_id field and filtering is automatically added
    # All queries automatically include user context

# Usage in agents - no special handling needed
class DocumentAgent(FiberAgent):
    def run_agent(self, input_data, current_user=None):
        # This query is automatically scoped to current user
        documents = Document.objects.all()  # Only returns current user's docs
        
        # Creating new documents automatically includes user_id
        new_doc = Document.objects.create(
            filename=input_data['filename'],
            content=input_data['content']
            # user_id is automatically set from current_user
        )
        
        return {'documents': len(documents), 'created': new_doc.id}

API-Level Isolation

# api_endpoints.py - Automatic user scoping
from fiberwise_sdk import UserScopedAPIView

class DocumentsAPI(UserScopedAPIView):
    """All endpoints automatically scoped to authenticated user"""
    
    def get(self, request):
        # Documents automatically filtered by user
        documents = Document.objects.all()
        return self.json_response([{
            'id': doc.id,
            'filename': doc.filename,
            'created_at': doc.created_at
        } for doc in documents])
    
    def post(self, request):
        # New documents automatically associated with user
        document = Document.objects.create(
            filename=request.data['filename'],
            content=request.data['content']
        )
        return self.json_response({
            'id': document.id,
            'message': 'Document created successfully'
        })
    
    def delete(self, request, document_id):
        # Can only delete own documents
        try:
            document = Document.objects.get(id=document_id)
            document.delete()
            return self.json_response({'message': 'Document deleted'})
        except Document.DoesNotExist:
            # Returns 404 if document doesn't exist OR belongs to different user
            return self.not_found_response()

# Routes are automatically protected
# GET /api/documents - only shows current user's documents
# POST /api/documents - creates document for current user
# DELETE /api/documents/123 - only deletes if user owns document

File System Isolation

Storage Path Structure

# Automatic user-based file organization
/storage/
├── users/
│   ├── user_a1b2c3d4/          # User 1's isolated directory
│   │   ├── uploads/
│   │   ├── generated/
│   │   ├── temp/
│   │   └── archives/
│   ├── user_e5f6g7h8/          # User 2's isolated directory
│   │   ├── uploads/
│   │   ├── generated/
│   │   ├── temp/
│   │   └── archives/
│   └── user_i9j0k1l2/          # User 3's isolated directory
│       ├── uploads/
│       ├── generated/
│       ├── temp/
│       └── archives/
└── shared/                     # System-level shared resources
    ├── templates/
    ├── public_assets/
    └── system_files/

Secure File Access

# storage_service.py - User-isolated file operations
class UserStorageService:
    def __init__(self, current_user):
        self.current_user = current_user
        self.user_path = f"users/{current_user.id}"
    
    def upload_file(self, file_content, filename, folder="uploads"):
        """Upload file to user's isolated directory"""
        file_path = f"{self.user_path}/{folder}/{filename}"
        
        # Ensure user can only upload to their directory
        if not file_path.startswith(self.user_path):
            raise SecurityError("Access denied: Path outside user directory")
        
        return self.storage_backend.save(file_path, file_content)
    
    def get_file(self, file_id):
        """Get file - automatically checks ownership"""
        file_record = FileRecord.objects.get(
            id=file_id,
            user_id=self.current_user.id  # Automatic user filter
        )
        
        if not file_record:
            raise FileNotFoundError("File not found or access denied")
        
        return self.storage_backend.load(file_record.file_path)
    
    def list_files(self, folder=""):
        """List files in user's directory only"""
        user_folder = f"{self.user_path}/{folder}"
        
        # Only list files within user's directory
        files = self.storage_backend.list_files(user_folder)
        
        return [{
            'id': file.id,
            'filename': file.filename,
            'size': file.size,
            'created_at': file.created_at
        } for file in files]
    
    def generate_secure_url(self, file_id, expires_in=3600):
        """Generate secure URL for user's file only"""
        file_record = FileRecord.objects.get(
            id=file_id,
            user_id=self.current_user.id
        )
        
        if not file_record:
            raise PermissionError("File not found or access denied")
        
        return self.storage_backend.generate_signed_url(
            file_record.file_path,
            expires_in=expires_in
        )

Agent Execution Isolation

User Context in Agents

# content_processor.py
from fiberwise_sdk import FiberAgent

class ContentProcessor(FiberAgent):
    """Agent that processes user content with automatic isolation"""
    
    def run_agent(self, input_data, current_user=None, storage_service=None):
        # current_user is automatically injected with authenticated user context
        # storage_service is automatically scoped to current user
        
        user_id = current_user.id
        user_name = current_user.name
        
        # All database operations automatically scoped to current user
        user_documents = Document.objects.all()  # Only current user's documents
        
        # File operations automatically isolated to user's directory
        user_files = storage_service.list_files("documents")
        
        # Process content for this specific user
        result = self._process_user_content(user_documents, user_files)
        
        # Save results - automatically associated with current user
        processed_doc = ProcessedDocument.objects.create(
            title=f"Processed content for {user_name}",
            content=result,
            # user_id automatically set from current_user
        )
        
        return {
            'status': 'success',
            'user_id': user_id,
            'documents_processed': len(user_documents),
            'files_processed': len(user_files),
            'result_document_id': processed_doc.id
        }
    
    def _process_user_content(self, documents, files):
        # Process content with user-specific context
        # All data is already isolated to the current user
        return "Processed content based on user's data"

User-Isolated Agent Deployment

# user_isolated_agent.py
class UserIsolatedAnalytics(FiberAgent):
    """Agent that provides analytics while maintaining user isolation"""
    
    def run_agent(self, input_data, current_user=None):
        # Each user gets their own analytics, completely isolated
        
        # User-specific metrics
        user_metrics = {
            'documents_created': Document.objects.count(),  # Only user's docs
            'storage_used': self._calculate_storage_usage(current_user.id),
            'agents_executed': self._get_activation_count(current_user.id),
            'last_activity': self._get_last_activity(current_user.id)
        }
        
        # Industry benchmarks (anonymized, aggregated data)
        industry_benchmarks = self._get_anonymized_benchmarks()
        
        return {
            'user_metrics': user_metrics,
            'industry_benchmarks': industry_benchmarks,
            'privacy_note': 'Your data is completely isolated and private'
        }
    
    def _get_anonymized_benchmarks(self):
        """Get industry averages without exposing individual user data"""
        # This would use aggregated, anonymized data
        return {
            'average_documents_per_user': 45,
            'average_monthly_activations': 120,
            'typical_storage_usage': '2.3 GB'
        }

Administrative Access & Overrides

Controlled Admin Access

# admin_access.py
from fiberwise_sdk import AdminRequiredMixin

class AdminUserDataAccess(AdminRequiredMixin):
    """Controlled admin access to user data with full audit logging"""
    
    def access_user_data(self, target_user_id, reason, current_admin):
        # Verify admin permissions
        if not current_admin.has_permission('admin.access_user_data'):
            raise PermissionError("Admin access denied")
        
        # Log admin access for audit trail
        AdminAccessLog.objects.create(
            admin_user=current_admin,
            target_user_id=target_user_id,
            access_reason=reason,
            access_type='user_data_access',
            timestamp=datetime.now(),
            ip_address=self.request.remote_addr
        )
        
        # Override user isolation for admin access
        with admin_context(current_admin):
            user_documents = Document.objects.filter(user_id=target_user_id)
            user_files = FileRecord.objects.filter(user_id=target_user_id)
            
            return {
                'user_id': target_user_id,
                'documents': list(user_documents.values()),
                'files': list(user_files.values()),
                'access_logged': True,
                'accessed_by': current_admin.id
            }
    
    def bulk_user_operation(self, operation, user_ids, current_admin):
        """Perform bulk operations across multiple users (admin only)"""
        if not current_admin.has_permission('admin.bulk_operations'):
            raise PermissionError("Bulk operations not permitted")
        
        results = []
        for user_id in user_ids:
            try:
                with admin_context(current_admin):
                    if operation == 'export_data':
                        result = self._export_user_data(user_id)
                    elif operation == 'cleanup_temp_files':
                        result = self._cleanup_temp_files(user_id)
                    else:
                        raise ValueError(f"Unknown operation: {operation}")
                    
                    results.append({
                        'user_id': user_id,
                        'status': 'success',
                        'result': result
                    })
                    
                    # Log each operation
                    AdminAccessLog.objects.create(
                        admin_user=current_admin,
                        target_user_id=user_id,
                        access_reason=f"Bulk operation: {operation}",
                        access_type='bulk_operation'
                    )
                    
            except Exception as e:
                results.append({
                    'user_id': user_id,
                    'status': 'error',
                    'error': str(e)
                })
        
        return {
            'operation': operation,
            'total_users': len(user_ids),
            'successful': len([r for r in results if r['status'] == 'success']),
            'failed': len([r for r in results if r['status'] == 'error']),
            'results': results
        }

Audit & Compliance Logging

# audit_logging.py
class UserIsolationAuditLog(models.Model):
    """Comprehensive audit logging for user isolation events"""
    
    timestamp = models.DateTimeField(auto_now_add=True)
    user_id = models.UUIDField()
    action = models.CharField(max_length=100)  # 'data_access', 'file_upload', etc.
    resource_type = models.CharField(max_length=50)  # 'document', 'file', 'agent'
    resource_id = models.CharField(max_length=100)
    ip_address = models.GenericIPAddressField()
    user_agent = models.TextField()
    
    # For admin access events
    admin_user_id = models.UUIDField(null=True, blank=True)
    admin_reason = models.TextField(null=True, blank=True)
    
    # Access control results
    access_granted = models.BooleanField()
    denial_reason = models.TextField(null=True, blank=True)
    
    class Meta:
        indexes = [
            models.Index(fields=['user_id', 'timestamp']),
            models.Index(fields=['admin_user_id', 'timestamp']),
            models.Index(fields=['action', 'timestamp'])
        ]

def log_isolation_event(user_id, action, resource_type, resource_id, 
                       access_granted, request, admin_user=None, reason=None):
    """Log user isolation events for audit and compliance"""
    UserIsolationAuditLog.objects.create(
        user_id=user_id,
        action=action,
        resource_type=resource_type,
        resource_id=resource_id,
        ip_address=get_client_ip(request),
        user_agent=request.META.get('HTTP_USER_AGENT', ''),
        admin_user_id=admin_user.id if admin_user else None,
        admin_reason=reason,
        access_granted=access_granted,
        denial_reason=reason if not access_granted else None
    )

Best Practices & Security Guidelines

Monitoring & Alerting

# security_monitoring.py
class UserIsolationMonitor:
    """Monitor and alert on isolation security events"""
    
    def monitor_cross_user_access_attempts(self):
        """Detect potential cross-user access attempts"""
        
        # Monitor failed access attempts
        recent_failures = UserIsolationAuditLog.objects.filter(
            access_granted=False,
            timestamp__gte=datetime.now() - timedelta(minutes=15)
        ).values('user_id').annotate(
            failure_count=Count('id')
        ).filter(failure_count__gt=10)
        
        for user_failures in recent_failures:
            self.alert_security_team(
                f"Multiple access failures for user {user_failures['user_id']}"
            )
    
    def monitor_admin_access_patterns(self):
        """Alert on unusual admin access patterns"""
        
        admin_accesses = AdminAccessLog.objects.filter(
            timestamp__gte=datetime.now() - timedelta(hours=24)
        ).values('admin_user_id').annotate(
            access_count=Count('id'),
            unique_users_accessed=Count('target_user_id', distinct=True)
        )
        
        for admin_access in admin_accesses:
            if admin_access['unique_users_accessed'] > 50:
                self.alert_security_team(
                    f"Admin {admin_access['admin_user_id']} accessed "
                    f"{admin_access['unique_users_accessed']} user accounts in 24h"
                )
    
    def validate_isolation_integrity(self):
        """Verify isolation is working correctly"""
        
        # Check for any data without proper user_id
        orphaned_documents = Document.objects.filter(user_id__isnull=True)
        if orphaned_documents.exists():
            self.alert_security_team(
                f"Found {orphaned_documents.count()} documents without user_id"
            )
        
        # Verify file system isolation
        self.audit_file_system_permissions()
    
    def alert_security_team(self, message):
        """Send security alerts to the team"""
        # Implementation would send alerts via email, Slack, etc.
        pass

Next Steps

Ready to implement secure user isolation? Explore these resources: