Server-Side Session Management

Overview

This document describes the server-side session management system implemented to address the security vulnerability where sessions were not being properly expired on the server after logout.

Problem Statement

The pen test identified that user sessions were only being invalidated client-side (by clearing cookies) but not server-side. This meant that if a session token was captured, it could potentially still be used even after the user logged out.

Solution

Implemented a comprehensive server-side session tracking system that:

1. Creates a session record in the database when a user logs in
2. Validates sessions on protected endpoints
3. Properly invalidates sessions on logout
4. Automatically expires old sessions
5. Provides session activity tracking

Architecture

Database Model

Table: or_user_sessions

Column	Type	Description
id	Integer	Primary key
user_id	Integer	Foreign key to user table
session_token	String(255)	Unique session identifier
flask_session_id	String(255)	Flask session ID (if available)
ip_address	String(45)	Client IP address
user_agent	Text	Client user agent string
created_dts	DateTime	Session creation timestamp
last_activity_dts	DateTime	Last activity timestamp
expires_dts	DateTime	Session expiration timestamp
is_active	Boolean	Whether session is active
invalidated_dts	DateTime	When session was invalidated
invalidation_reason	String(100)	Reason for invalidation (logout, expired, security)

Model Location: app/api/user/models.py - UserSession class

Session Service

Location: app/api/user/services/user_session_service.py

Key Methods:

• create_session(user_id, duration_hours) - Create a new session
• get_session_by_token(session_token) - Retrieve a session
• get_active_session_by_token(session_token) - Get active session
• invalidate_session(session_token, reason) - Invalidate a session
• invalidate_user_sessions(user_id, reason, exclude_session_token) - Invalidate all user sessions
• validate_and_update_session(session_token) - Validate and update activity
• cleanup_expired_sessions(days_old) - Remove old sessions

Login Flow

Location: app/api/auth/controllers/login_controller.py

When a user logs in successfully:

1. Flask-Login session is created (existing behavior)
2. A new UserSession record is created with:
   • Unique session token (32-byte URL-safe token)
   • User ID
   • IP address and user agent
   • Expiration time (default: 720 hours / 30 days)
3. Session token is stored in Flask session: session['user_session_token']
4. Session details are logged

Logout Flow

Location: app/api/auth/controllers/logout_controller.py

When a user logs out:

1. Retrieve session token from Flask session
2. Call UserSessionService.invalidate_session() with reason="logout"
3. Session is marked as is_active=False with invalidation timestamp
4. Flask session is cleared (existing behavior)
5. Flask-Login logout is called (existing behavior)
6. Session cookie is expired (existing behavior)

Session Validation

Location: app/api/utils/decorators/session_validator.py

Decorator: @validate_session

Apply this decorator to protected endpoints to validate server-side sessions:

from app.api.utils.decorators.session_validator import validate_session
from flask_login import login_required

@app.route('/protected')
@login_required
@validate_session
def protected_endpoint():
    # Your code here
    pass

Validation Flow:

1. Check if user is authenticated (via Flask-Login)
2. Retrieve session token from Flask session
3. Validate session is active and not expired
4. Update last activity timestamp
5. If invalid: log user out and return 401 error
6. If valid: continue to endpoint

Backward Compatibility:

• If no session token exists (legacy session), a warning is logged but request continues
• This ensures existing sessions aren't broken during rollout

Session Cleanup

Location: app/api/user/jobs/cleanup_expired_sessions.py

A scheduled job to clean up old sessions from the database.

Configuration:

• Default: Remove sessions older than 30 days
• Run frequency: Recommended daily

Manual Execution:

python -m app.api.user.jobs.cleanup_expired_sessions

Scheduled Job: Configure via cron or EasyCron to run daily:

0 2 * * * cd /path/to/app && python -m app.api.user.jobs.cleanup_expired_sessions

Database Migration

File: migrations/versions/2025_12_31_add_user_sessions_table.py

To Apply:

flask db upgrade

Migration includes:

• Table creation with proper constraints
• Indexes for performance optimization
• Sequence for auto-incrementing ID
• Composite index for common queries (user_id + is_active + expires_dts)

Usage Examples

Creating a Session (Login)

from app.api.user.services.user_session_service import UserSessionService

# After successful authentication
user_session = UserSessionService.create_session(
    user_id=user.id,
    duration_hours=720  # 30 days
)

# Store token in Flask session
session['user_session_token'] = user_session.session_token

Invalidating a Session (Logout)

from app.api.user.services.user_session_service import UserSessionService

session_token = session.get('user_session_token')
if session_token:
    UserSessionService.invalidate_session(session_token, reason='logout')

Invalidating All User Sessions (Security Event)

from app.api.user.services.user_session_service import UserSessionService

# Invalidate all sessions except current one
UserSessionService.invalidate_user_sessions(
    user_id=user.id,
    reason='password_changed',
    exclude_session_token=current_session_token
)

Checking Active Sessions

from app.api.user.services.user_session_service import UserSessionService

active_sessions = UserSessionService.get_user_active_sessions(user_id=user.id)
print(f"User has {len(active_sessions)} active sessions")

Security Considerations

Session Token Generation

• Uses Python's secrets.token_urlsafe(32) for cryptographically secure random tokens
• 32 bytes = 256 bits of entropy
• URL-safe base64 encoding (43 characters)

Session Expiration

• Default: 720 hours (30 days) to match Flask-Login duration
• Configurable per login
• Both client (cookie) and server (database) expiration

Activity Tracking

• last_activity_dts updated on each validated request
• Enables detection of inactive sessions
• Can be used for "idle timeout" feature in future

Audit Trail

• All sessions tracked with creation time, IP, user agent
• Invalidation reason recorded (logout, expired, security)
• Useful for security audits and forensics

Performance Considerations

Database Indexes

• user_id - Fast lookup of user's sessions
• session_token - Unique index for fast validation
• is_active - Filter active sessions
• expires_dts - Find expired sessions
• Composite index (user_id, is_active, expires_dts) - Optimized for common queries

Cleanup Job

• Prevents table bloat by removing old sessions
• Recommended to run daily during low-traffic hours
• Configurable retention period (default 30 days)

Query Optimization

• Uses SQLAlchemy select() statements with proper filtering
• Avoids loading full objects when not needed
• Batch operations for invalidation

Testing

Unit Tests

Test the session service methods:

def test_create_session():
    session = UserSessionService.create_session(user_id=1)
    assert session.is_valid()
    assert session.user_id == 1

def test_invalidate_session():
    session = UserSessionService.create_session(user_id=1)
    token = session.session_token

    UserSessionService.invalidate_session(token, reason='test')

    retrieved = UserSessionService.get_session_by_token(token)
    assert not retrieved.is_valid()
    assert retrieved.invalidation_reason == 'test'

Integration Tests

Test the full login/logout flow:

def test_login_creates_session(client):
    response = client.post('/api/auth/login', data={
        'email_address': 'test@example.com',
        'password': 'password'
    })

    assert response.status_code == 200
    assert 'user_session_token' in session

def test_logout_invalidates_session(client):
    # Login first
    client.post('/api/auth/login', data=...)
    token = session['user_session_token']

    # Logout
    client.post('/api/auth/logout')

    # Verify session is invalidated
    user_session = UserSessionService.get_session_by_token(token)
    assert not user_session.is_valid()

Monitoring and Alerts

Metrics to Monitor

• Number of active sessions per user
• Session creation rate
• Session invalidation reasons distribution
• Failed session validation attempts
• Session table size growth

Recommended Alerts

• Alert if user has > 10 concurrent sessions (possible account compromise)
• Alert on high rate of failed session validations (possible attack)
• Alert if session cleanup job fails
• Alert if session table size exceeds threshold

Future Enhancements

Possible Improvements

1. Idle Timeout - Auto-invalidate sessions after X minutes of inactivity
2. Device Management - Allow users to view and revoke sessions by device
3. Geographic Anomaly Detection - Alert on login from unusual location
4. Session Limiting - Limit number of concurrent sessions per user
5. Remember Me - Separate long-lived "remember me" tokens
6. Two-Factor Session Binding - Require 2FA for sensitive operations even with valid session

Rollout Plan

Phase 1: Database Setup (Completed)

• ✅ Create migration
• ✅ Apply to development environment
• ✅ Test migration rollback

Phase 2: Core Implementation (Completed)

• ✅ Implement session model
• ✅ Implement session service
• ✅ Update login controller
• ✅ Update logout controller

Phase 3: Validation (Completed)

• ✅ Create session validator decorator
• ✅ Add to critical endpoints

Phase 4: Cleanup (Completed)

• ✅ Create cleanup job
• ✅ Schedule in production

Phase 5: Deployment (Next Steps)

1. Deploy to staging environment
2. Test all authentication flows
3. Monitor session creation/invalidation
4. Deploy to production (during low-traffic window)
5. Monitor for issues
6. Gradually add @validate_session to more endpoints

Phase 6: Enforcement (Future)

1. Add @validate_session to all protected routes
2. Remove backward compatibility warning
3. Require valid session for all authenticated requests

Troubleshooting

Issue: Session validation failing for all users

Cause: Migration not applied or service not creating sessions Solution:

1. Check if migration has been applied: flask db current
2. Check logs for session creation errors
3. Verify user_session_token is being stored in Flask session

Issue: Users logged out unexpectedly

Cause: Session expiration time too short or validation too strict Solution:

1. Check session expiration settings (default 720 hours)
2. Review session validation logs
3. Ensure cleanup job isn't removing active sessions

Issue: Session table growing too large

Cause: Cleanup job not running or configured with too long retention Solution:

1. Verify cleanup job is scheduled and running
2. Reduce retention period if needed
3. Manually run cleanup: python -m app.api.user.jobs.cleanup_expired_sessions

Support

For questions or issues related to session management:

• Review this documentation
• Check logs for session-related errors
• Review session records in or_user_sessions table
• Contact the security team for security-related questions

Changelog

2025-12-31 - Initial Implementation

• Created UserSession model and or_user_sessions table
• Implemented UserSessionService with full lifecycle management
• Updated login/logout controllers to manage sessions
• Created @validate_session decorator
• Added cleanup job for expired sessions
• Created comprehensive documentation