Security Improvements: Organization Filtering

How we tightened org-scoped queries and added regression coverage so one tenant cannot read or mutate another’s rows through ordinary code paths.

Overview

• Tightened org filters on tasks, CRUD, users, permissions, status utilities, and stats.
• Regression tests in tests/backend/test_security_fixes.py and optional pytest -m security.
• Optional script scripts/check_organization_filtering.py for CI; experimental org middleware exists behind flags.

Critical Security Fixes Implemented

1. Task Management Security

Files: app/services/task_management.py, app/crud.py Issue: Task queries lacked organization filtering, allowing cross-tenant access Fix: Added organization_id parameters and filtering to all task operations

# Before (VULNERABLE)
task = db.query(models.Task).filter(models.Task.id == task_id).first()

# After (SECURE)
def get_task(db: Session, task_id: UUID, organization_id: str = None) -> Optional[models.Task]:
    query = db.query(models.Task).filter(models.Task.id == task_id)
    if organization_id:
        query = query.filter(models.Task.organization_id == UUID(organization_id))
    return query.first()

2. CRUD Operations Security

Files: app/crud.py Issue: remove_tag, task queries without organization filtering Fix: Added organization filtering to prevent cross-tenant tag manipulation

# Before (VULNERABLE)
db_tag = db.query(models.Tag).filter(models.Tag.id == tag_id).first()

# After (SECURE)
def remove_tag(db: Session, tag_id: UUID, entity_id: UUID, entity_type: str, organization_id: str = None):
    tag_query = db.query(models.Tag).filter(models.Tag.id == tag_id)
    if organization_id:
        tag_query = tag_query.filter(models.Tag.organization_id == UUID(organization_id))
    # ... rest of function

3. User Router Security

Files: app/routers/user.py Issue: User update queries without organization filtering Fix: Added organization filtering with superuser exceptions

# Before (VULNERABLE)
db_user = db.query(User).filter(User.id == user_id).first()

# After (SECURE)
user_query = db.query(User).filter(User.id == user_id)
if not current_user.is_superuser and current_user.organization_id:
    user_query = user_query.filter(User.organization_id == current_user.organization_id)
db_user = user_query.first()

4. Auth Permissions Security

Files: app/auth/permissions.py Issue: Resource permission checks without organization filtering Fix: Applied organization filtering before permission validation

5. Status Utility Security

Files: app/utils/status.py Issue: Status queries without organization filtering Fix: Added organization-aware status creation and lookup

6. Statistics Security

Files: app/services/stats/ Issue: Statistics queries without organization filtering Fix: Added organization context to StatsCalculator constructor

Security Test Suite

Comprehensive Test Coverage

File: tests/backend/test_security_fixes.py

The security test suite includes:

• Cross-tenant access prevention tests for all fixed vulnerabilities
• Organization filtering validation for CRUD operations
• Auth permissions security tests
• Regression tests to prevent future vulnerabilities
• Security markers for targeted test execution

# Run all security tests
pytest tests/backend/test_security_fixes.py -v

# Run only security-marked tests
pytest -m security

CI/CD Security Integration

Automated Security Scanning

File: scripts/check_organization_filtering.py

The security check script automatically scans the codebase for:

• Database queries missing organization filtering
• HIGH severity issues (queries on organization-aware models)
• MEDIUM severity issues (potentially unsafe queries)

# Run security check
python scripts/check_organization_filtering.py --verbose

# Setup CI/CD integration
python scripts/check_organization_filtering.py --setup-ci

GitHub Actions Integration

The script can generate GitHub Actions workflows for:

• Pull request security checks
• Automated security test execution
• Security issue reporting in PR comments

Query-Level Organization Filtering Middleware

Experimental Middleware Solution

File: app/middleware/organization_filter.py

Provides automatic organization filtering through:

1. Context Manager Approach (Recommended)

with with_organization_context("org-123"):
    tests = db.query(Test).all()  # Automatically filtered

2. Organization-Aware Session Wrapper (Recommended)

org_session = get_organization_aware_session(db, "org-123")
tests = org_session.query(Test).all()  # Automatically filtered

3. Decorator Approach

@organization_aware_query
def get_user_tests(db: Session, user_id: str, organization_id: str):
    return db.query(Test).filter(Test.user_id == user_id).all()

Safety Features

• Disabled by default for safety
• Bypass mechanisms for administrative operations
• Comprehensive logging for monitoring
• Query interception with automatic filtering

Security Best Practices

1. Understanding Query Safety Levels

Safe (filter often optional):

# ID-based queries are SAFE - UUIDs are globally unique
def get_entity_by_id(db: Session, entity_id: UUID) -> Optional[Entity]:
    return db.query(Entity).filter(Entity.id == entity_id).first()

# Primary key lookups are SAFE
entity = db.query(Entity).get(entity_id)

# User queries (handled separately)
user = db.query(User).filter(User.id == user_id).first()

Require organization_id (list/search and similar):

# List queries without filters - DANGEROUS
def get_all_entities(db: Session, organization_id: str) -> List[Entity]:
    return db.query(Entity).filter(
        Entity.organization_id == UUID(organization_id)  # REQUIRED!
    ).all()

# Search by non-unique fields - DANGEROUS
def get_entities_by_name(db: Session, name: str, organization_id: str) -> List[Entity]:
    return db.query(Entity).filter(
        Entity.name == name,
        Entity.organization_id == UUID(organization_id)  # REQUIRED!
    ).all()

2. Direct Parameter Passing (Current Approach)

# RECOMMENDED: Explicit organization filtering for list/search queries
def get_entities(db: Session, organization_id: str) -> List[Entity]:
    return db.query(Entity).filter(
        Entity.organization_id == UUID(organization_id)
    ).all()

3. Always Validate Organization Context

# Verify organization context is provided
if not organization_id:
    raise ValueError("Organization context required for multi-tenant operation")

4. Use Security Tests

# Test cross-tenant access prevention
def test_cross_tenant_prevention(self, test_db: Session):
    # Create entities in different organizations
    # Verify org1 user cannot access org2 data

5. Apply Defense in Depth

Combine explicit org filters in app code, FK integrity, optional middleware, and tests.

Performance Considerations

Query Performance

• Indexed organization_id fields ensure fast filtering
• Composite indexes on (organization_id, other_fields) for complex queries
• Query plan analysis to verify efficient execution

Security vs. Performance Trade-offs

• Direct parameter passing: Best performance, explicit security
• Middleware solutions: Slight overhead, automatic security
• Choose based on: Team expertise, maintenance requirements, performance needs

Future work

Finish stats coverage, keep chipping away at script findings, and evolve middleware only where it clearly reduces missed filters.