Access Control Remediation

# ❌ VULNERABLE - No ownership check
@app.route('/api/documents/<int:doc_id>')
def get_document(doc_id):
    document = Document.query.get_or_404(doc_id)
    return jsonify(document.to_dict())

# ✅ SECURE - Ownership verification
@app.route('/api/documents/<int:doc_id>')
@login_required
def get_document(doc_id):
    document = Document.query.filter_by(
        id=doc_id,
        owner_id=current_user.id  # Verify ownership
    ).first_or_404()
    return jsonify(document.to_dict())

# ✅ SECURE - With shared access support
@app.route('/api/documents/<int:doc_id>')
@login_required
def get_document(doc_id):
    document = Document.query.get_or_404(doc_id)
    
    # Check if user has any form of access
    if not has_document_access(current_user, document):
        abort(403)
    
    return jsonify(document.to_dict())

def has_document_access(user, document):
    """Check if user can access document"""
    # Owner always has access
    if document.owner_id == user.id:
        return True
    
    # Check if document is shared with user
    share = DocumentShare.query.filter_by(
        document_id=document.id,
        user_id=user.id
    ).first()
    
    return share is not None

from functools import wraps
from flask import abort

# Define permissions
class Permission:
    READ = 'read'
    WRITE = 'write'
    DELETE = 'delete'
    ADMIN = 'admin'

# Role definitions
ROLE_PERMISSIONS = {
    'viewer': [Permission.READ],
    'editor': [Permission.READ, Permission.WRITE],
    'admin': [Permission.READ, Permission.WRITE, Permission.DELETE, Permission.ADMIN],
}

def requires_permission(permission):
    """Decorator to check permissions"""
    def decorator(f):
        @wraps(f)
        def decorated_function(*args, **kwargs):
            if not current_user.is_authenticated:
                abort(401)
            
            user_permissions = ROLE_PERMISSIONS.get(current_user.role, [])
            if permission not in user_permissions:
                abort(403)
            
            return f(*args, **kwargs)
        return decorated_function
    return decorator

# Usage
@app.route('/api/users', methods=['DELETE'])
@requires_permission(Permission.ADMIN)
def delete_user():
    # Only admins can reach here
    pass

@app.route('/api/documents', methods=['POST'])
@requires_permission(Permission.WRITE)
def create_document():
    # Editors and admins can create
    pass

from dataclasses import dataclass
from typing import Any, Dict

@dataclass
class AccessRequest:
    subject: Dict[str, Any]    # User attributes
    resource: Dict[str, Any]   # Resource attributes  
    action: str                # Requested action
    environment: Dict[str, Any] = None  # Context (time, IP, etc.)

class PolicyEngine:
    def evaluate(self, request: AccessRequest) -> bool:
        """Evaluate access request against policies"""
        
        # Policy 1: Users can only access resources in their department
        if request.resource.get('department') != request.subject.get('department'):
            # Exception for admins
            if 'admin' not in request.subject.get('roles', []):
                return False
        
        # Policy 2: Sensitive documents require manager role
        if request.resource.get('classification') == 'sensitive':
            if 'manager' not in request.subject.get('roles', []):
                return False
        
        # Policy 3: Write access only during business hours
        if request.action in ['write', 'delete']:
            hour = request.environment.get('hour', 12)
            if hour < 9 or hour > 17:
                if 'admin' not in request.subject.get('roles', []):
                    return False
        
        # Policy 4: Contractors cannot delete
        if request.action == 'delete':
            if request.subject.get('user_type') == 'contractor':
                return False
        
        return True

# Usage
policy = PolicyEngine()

def check_access(user, resource, action):
    request = AccessRequest(
        subject={
            'id': user.id,
            'department': user.department,
            'roles': user.roles,
            'user_type': user.user_type,
        },
        resource={
            'id': resource.id,
            'department': resource.department,
            'classification': resource.classification,
        },
        action=action,
        environment={
            'hour': datetime.now().hour,
            'ip': request.remote_addr,
        }
    )
    return policy.evaluate(request)

const express = require('express');

// Authorization middleware
const authorize = (...allowedRoles) => {
  return (req, res, next) => {
    if (!req.user) {
      return res.status(401).json({ error: 'Authentication required' });
    }
    
    if (!allowedRoles.includes(req.user.role)) {
      return res.status(403).json({ error: 'Insufficient permissions' });
    }
    
    next();
  };
};

// Resource ownership middleware
const ownsResource = (resourceGetter) => {
  return async (req, res, next) => {
    const resource = await resourceGetter(req);
    
    if (!resource) {
      return res.status(404).json({ error: 'Resource not found' });
    }
    
    // Admin bypass
    if (req.user.role === 'admin') {
      req.resource = resource;
      return next();
    }
    
    // Ownership check
    if (resource.ownerId !== req.user.id) {
      return res.status(403).json({ error: 'Access denied' });
    }
    
    req.resource = resource;
    next();
  };
};

// Usage
app.get('/api/admin/users', 
  authorize('admin'),
  adminController.listUsers
);

app.get('/api/documents/:id',
  authorize('user', 'admin'),
  ownsResource(req => Document.findById(req.params.id)),
  documentController.getDocument
);

app.delete('/api/documents/:id',
  authorize('user', 'admin'),
  ownsResource(req => Document.findById(req.params.id)),
  documentController.deleteDocument
);

import pytest

class TestDocumentAccessControl:
    """Test IDOR and access control vulnerabilities"""
    
    def test_user_cannot_access_others_document(self, client):
        """Horizontal privilege escalation test"""
        # User A creates document
        user_a = create_user()
        doc = create_document(owner=user_a)
        
        # User B tries to access
        user_b = create_user()
        client.login(user_b)
        
        response = client.get(f'/api/documents/{doc.id}')
        assert response.status_code == 403
    
    def test_user_cannot_delete_others_document(self, client):
        """Test IDOR on destructive actions"""
        user_a = create_user()
        doc = create_document(owner=user_a)
        
        user_b = create_user()
        client.login(user_b)
        
        response = client.delete(f'/api/documents/{doc.id}')
        assert response.status_code == 403
        assert Document.query.get(doc.id) is not None  # Not deleted
    
    def test_regular_user_cannot_access_admin_endpoint(self, client):
        """Vertical privilege escalation test"""
        user = create_user(role='user')
        client.login(user)
        
        response = client.get('/api/admin/users')
        assert response.status_code == 403
    
    def test_parameter_tampering_blocked(self, client):
        """Test role parameter tampering"""
        user = create_user(role='user')
        client.login(user)
        
        # Try to elevate own role
        response = client.put('/api/profile', json={'role': 'admin'})
        
        user.refresh()
        assert user.role == 'user'  # Role unchanged