"""Tenancy helpers — cross-tenant access control (IDOR fix, July 2026).
Every route that reads a ``company_id`` from the request (query string,
URL param, or JSON body) MUST verify the current user is a member of that
company before touching company-scoped data. These helpers centralize
that check.
Usage patterns
--------------
1. Query-string based pages (dashboard views)::
company_id = resolve_company_id(default=companies[0].id)
Returns a company_id the current user is guaranteed to belong to,
or aborts with 403 if the requested company_id is not accessible.
2. URL-param API routes::
@dashboard_bp.route('/api/kpis/<company_id>')
@login_required
@require_company_access
def api_kpis(company_id):
...
Super admins (User.role == 'super_admin') bypass membership checks.
"""
from functools import wraps
from flask import request, jsonify, abort
from flask_login import current_user
from ..models import UserCompany
def user_in_company(user_id, company_id):
"""Return True if the user is a member of the company."""
if not user_id or not company_id:
return False
return UserCompany.query.filter_by(
user_id=user_id,
company_id=company_id,
).first() is not None
def _is_super_admin():
return (
current_user.is_authenticated
and getattr(current_user, 'role', None) == 'super_admin'
)
def resolve_company_id(default=None):
"""Resolve a company_id from ``request.args`` with a membership check.
- If no ``company_id`` query param is present, returns ``default``.
- If one is present and the current user is a member (or super admin),
returns it.
- Otherwise aborts with 403.
"""
company_id = request.args.get('company_id')
if not company_id:
return default
if _is_super_admin() or user_in_company(current_user.id, company_id):
return company_id
abort(403)
def require_company_access(f):
"""Decorator for routes with a ``company_id`` URL/query/body param.
Verifies the current user is a member of the company (super admins
bypass). Returns JSON 403 on failure — intended for API routes.
"""
@wraps(f)
def decorated(*args, **kwargs):
if not current_user.is_authenticated:
return jsonify({'error': 'Authentication required'}), 401
company_id = kwargs.get('company_id') or request.args.get('company_id')
if not company_id and request.is_json:
body = request.get_json(silent=True) or {}
company_id = body.get('company_id')
if not company_id:
return jsonify({'error': 'company_id required'}), 400
if not _is_super_admin() and not user_in_company(current_user.id, company_id):
return jsonify({'error': 'Forbidden — no access to this company'}), 403
return f(*args, **kwargs)
return decorated