Add Entra ID authentication with role-based access control
Implement Microsoft Entra ID (Azure AD) authentication with user and admin roles. Features: - OAuth 2.0 authentication flow using MSAL - User and admin role-based access control - Tenant ownership and access management - Server-side session storage - Protected routes with authentication decorators Authentication: - Users authenticate via Microsoft organizational accounts - Admin role automatically assigned to DEFAULT_ADMIN_EMAIL - User profiles synced from Microsoft Graph API Access Control: - Admins: Full access to all tenants and server settings - Users: Can create and manage their own tenants only - Tenant association: Auto-created when user adds products - API endpoints remain open for AR application access Database Changes: - New users table with email, azure_oid, name, and role - New user_tenants table for many-to-many tenant ownership - Case-insensitive admin email comparison Configuration: - Environment-based config via .env file - Azure AD credentials (client ID, secret, tenant ID) - Configurable redirect URI and admin email - Updated to Flask 2.2.5 for compatibility
This commit is contained in:
@@ -1,5 +1,6 @@
|
||||
import os
|
||||
from flask import Flask, redirect
|
||||
from flask_session import Session
|
||||
from werkzeug.routing import BaseConverter
|
||||
|
||||
def create_app(config_name='development'):
|
||||
@@ -12,6 +13,10 @@ def create_app(config_name='development'):
|
||||
|
||||
# Ensure data folder exists
|
||||
os.makedirs(app.config['DATA_FOLDER'], exist_ok=True)
|
||||
os.makedirs(app.config['SESSION_FILE_DIR'], exist_ok=True)
|
||||
|
||||
# Initialize Flask-Session for server-side sessions
|
||||
Session(app)
|
||||
|
||||
# Custom converter for tenant IDs
|
||||
class TenantConverter(BaseConverter):
|
||||
@@ -30,11 +35,17 @@ def create_app(config_name='development'):
|
||||
# Initialize database
|
||||
with app.app_context():
|
||||
from app.models.base import init_database
|
||||
from app.models.user import UserModel
|
||||
|
||||
init_database()
|
||||
# Create user tables
|
||||
UserModel.create_table()
|
||||
|
||||
# Register blueprints
|
||||
from app.blueprints import main_bp, tenant_bp, admin_bp, api_bp
|
||||
from app.blueprints.auth import auth_bp
|
||||
|
||||
app.register_blueprint(auth_bp)
|
||||
app.register_blueprint(main_bp)
|
||||
app.register_blueprint(tenant_bp)
|
||||
app.register_blueprint(admin_bp)
|
||||
|
||||
@@ -1,9 +1,11 @@
|
||||
from flask import render_template, request, redirect, url_for, flash, jsonify, current_app
|
||||
from flask import render_template, request, redirect, url_for, flash, jsonify, current_app, session
|
||||
from . import admin_bp
|
||||
from app.models import TenantModel, ProductModel, ARFieldModel
|
||||
from app.models import TenantModel, ProductModel, ARFieldModel, UserModel
|
||||
from app.services import ProductService
|
||||
from app.decorators.auth import tenant_access_required
|
||||
|
||||
@admin_bp.route('/add', methods=['GET', 'POST'])
|
||||
@tenant_access_required
|
||||
def add_product(tenant_id):
|
||||
"""Add a new product"""
|
||||
custom_fields = ARFieldModel.get_all(tenant_id)
|
||||
@@ -128,12 +130,18 @@ def add_product(tenant_id):
|
||||
for img in images_to_save:
|
||||
ProductModel.save_image(product_id, tenant_id, img['field_name'], img['data'], img['mime_type'])
|
||||
|
||||
# Associate the tenant with the user if not already associated
|
||||
user_id = session['user']['id']
|
||||
if not UserModel.has_access_to_tenant(user_id, tenant_id):
|
||||
UserModel.add_tenant(user_id, tenant_id)
|
||||
|
||||
flash('Product added successfully!')
|
||||
return redirect(f'/{tenant_id}/')
|
||||
|
||||
return render_template('admin/add_product.html', tenant_id=tenant_id, custom_fields=custom_fields)
|
||||
|
||||
@admin_bp.route('/edit/<product_id>', methods=['GET', 'POST'])
|
||||
@tenant_access_required
|
||||
def edit_product(tenant_id, product_id):
|
||||
"""Edit an existing product"""
|
||||
products = ProductModel.get_all(tenant_id)
|
||||
@@ -253,6 +261,7 @@ def edit_product(tenant_id, product_id):
|
||||
custom_fields=custom_fields)
|
||||
|
||||
@admin_bp.route('/delete/<product_id>', methods=['POST'])
|
||||
@tenant_access_required
|
||||
def delete_product(tenant_id, product_id):
|
||||
"""Delete a product"""
|
||||
products = ProductModel.get_all(tenant_id)
|
||||
@@ -266,6 +275,7 @@ def delete_product(tenant_id, product_id):
|
||||
return redirect(f'/{tenant_id}/')
|
||||
|
||||
@admin_bp.route('/generate_barcode/<product_id>/<code_type>')
|
||||
@tenant_access_required
|
||||
def generate_barcode(tenant_id, product_id, code_type):
|
||||
"""Redirect to the main barcode generation endpoint"""
|
||||
products = ProductModel.get_all(tenant_id)
|
||||
@@ -282,6 +292,7 @@ def generate_barcode(tenant_id, product_id, code_type):
|
||||
return redirect(f'/{tenant_id}/barcodes/{product_id}_{barcode_type}.png')
|
||||
|
||||
@admin_bp.route('/barcodes', methods=['GET'])
|
||||
@tenant_access_required
|
||||
def view_all_barcodes(tenant_id):
|
||||
"""Display all product barcodes for printing"""
|
||||
products = ProductModel.get_all(tenant_id)
|
||||
@@ -295,6 +306,7 @@ def view_all_barcodes(tenant_id):
|
||||
barcode_type=barcode_type)
|
||||
|
||||
@admin_bp.route('/ar_fields', methods=['GET', 'POST'])
|
||||
@tenant_access_required
|
||||
def manage_ar_fields(tenant_id):
|
||||
"""Manage custom AR content fields"""
|
||||
tenant = TenantModel.get_or_create(tenant_id)
|
||||
|
||||
6
src/app/blueprints/auth/__init__.py
Normal file
6
src/app/blueprints/auth/__init__.py
Normal file
@@ -0,0 +1,6 @@
|
||||
"""Authentication blueprint"""
|
||||
from flask import Blueprint
|
||||
|
||||
auth_bp = Blueprint('auth', __name__, url_prefix='/auth')
|
||||
|
||||
from . import routes
|
||||
122
src/app/blueprints/auth/routes.py
Normal file
122
src/app/blueprints/auth/routes.py
Normal file
@@ -0,0 +1,122 @@
|
||||
"""Authentication routes for Entra ID login/logout"""
|
||||
import secrets
|
||||
from flask import session, redirect, url_for, request, current_app, render_template, flash
|
||||
from . import auth_bp
|
||||
from app.services.msal_service import MSALService
|
||||
from app.models.user import UserModel
|
||||
|
||||
|
||||
@auth_bp.route('/login')
|
||||
def login():
|
||||
"""Initiate Entra ID login flow"""
|
||||
# Generate state token for CSRF protection
|
||||
state = secrets.token_urlsafe(16)
|
||||
session['auth_state'] = state
|
||||
|
||||
# Store the original requested URL to redirect after login
|
||||
if 'next' not in session and request.args.get('next'):
|
||||
session['next'] = request.args.get('next')
|
||||
|
||||
# Get authorization URL from MSAL
|
||||
auth_url = MSALService.get_auth_url(state=state)
|
||||
|
||||
return redirect(auth_url)
|
||||
|
||||
|
||||
@auth_bp.route('/callback')
|
||||
def callback():
|
||||
"""Handle the callback from Entra ID after authentication"""
|
||||
# Verify state to prevent CSRF
|
||||
state = request.args.get('state')
|
||||
if state != session.get('auth_state'):
|
||||
current_app.logger.error("State mismatch in auth callback - possible CSRF attack")
|
||||
flash('Authentication failed: Invalid state parameter', 'error')
|
||||
return redirect(url_for('main.index'))
|
||||
|
||||
# Clear the state
|
||||
session.pop('auth_state', None)
|
||||
|
||||
# Check for errors
|
||||
if 'error' in request.args:
|
||||
error = request.args.get('error')
|
||||
error_description = request.args.get('error_description', 'Unknown error')
|
||||
current_app.logger.error(f"Auth error: {error} - {error_description}")
|
||||
flash(f'Authentication failed: {error_description}', 'error')
|
||||
return redirect(url_for('main.index'))
|
||||
|
||||
# Get authorization code
|
||||
auth_code = request.args.get('code')
|
||||
if not auth_code:
|
||||
flash('Authentication failed: No authorization code received', 'error')
|
||||
return redirect(url_for('main.index'))
|
||||
|
||||
# Exchange code for token
|
||||
token_response = MSALService.acquire_token_by_auth_code(auth_code)
|
||||
|
||||
if 'error' in token_response:
|
||||
error = token_response.get('error')
|
||||
error_description = token_response.get('error_description', 'Unknown error')
|
||||
current_app.logger.error(f"Token error: {error} - {error_description}")
|
||||
flash(f'Authentication failed: {error_description}', 'error')
|
||||
return redirect(url_for('main.index'))
|
||||
|
||||
# Get user info from Microsoft Graph
|
||||
access_token = token_response.get('access_token')
|
||||
user_info = MSALService.get_user_info(access_token)
|
||||
|
||||
if not user_info:
|
||||
flash('Failed to retrieve user information', 'error')
|
||||
return redirect(url_for('main.index'))
|
||||
|
||||
# Get or create user in database
|
||||
user = UserModel.get_or_create_from_azure(
|
||||
email=user_info['email'],
|
||||
name=user_info['name'],
|
||||
azure_oid=user_info['azure_oid'],
|
||||
default_admin_email=current_app.config['DEFAULT_ADMIN_EMAIL']
|
||||
)
|
||||
|
||||
# Store user in session
|
||||
session['user'] = {
|
||||
'id': user['id'],
|
||||
'email': user['email'],
|
||||
'name': user['name'],
|
||||
'role': user['role']
|
||||
}
|
||||
|
||||
current_app.logger.info(f"User {user['email']} logged in successfully")
|
||||
|
||||
# Redirect to the originally requested page or home
|
||||
next_url = session.pop('next', None)
|
||||
if next_url:
|
||||
return redirect(next_url)
|
||||
|
||||
return redirect(url_for('main.index'))
|
||||
|
||||
|
||||
@auth_bp.route('/logout')
|
||||
def logout():
|
||||
"""Log out the current user"""
|
||||
if 'user' in session:
|
||||
user_email = session['user'].get('email')
|
||||
current_app.logger.info(f"User {user_email} logged out")
|
||||
|
||||
# Clear session
|
||||
session.clear()
|
||||
|
||||
# Remove from MSAL cache
|
||||
MSALService.remove_account()
|
||||
|
||||
# Redirect to Azure AD logout
|
||||
logout_url = (
|
||||
f"{current_app.config['AUTHORITY']}{current_app.config['AZURE_TENANT_ID']}/oauth2/v2.0/logout"
|
||||
f"?post_logout_redirect_uri={request.url_root}"
|
||||
)
|
||||
|
||||
return redirect(logout_url)
|
||||
|
||||
|
||||
@auth_bp.route('/unauthorized')
|
||||
def unauthorized():
|
||||
"""Show unauthorized access page"""
|
||||
return render_template('unauthorized.html'), 403
|
||||
@@ -1,17 +1,29 @@
|
||||
from flask import render_template, request, redirect, flash
|
||||
from flask import render_template, request, redirect, flash, session
|
||||
from . import main_bp
|
||||
from app.models import TenantModel, SettingsModel
|
||||
from app.models import TenantModel, SettingsModel, UserModel
|
||||
from app.decorators.auth import login_required, settings_access_required
|
||||
|
||||
@main_bp.route('/')
|
||||
@login_required
|
||||
def index():
|
||||
"""Tenant selection page"""
|
||||
tenants = TenantModel.get_all()
|
||||
user_id = session['user']['id']
|
||||
user = UserModel.get_by_id(user_id)
|
||||
|
||||
# Admins see all tenants, users only see their own
|
||||
if user['role'] == UserModel.ROLE_ADMIN:
|
||||
tenants = TenantModel.get_all()
|
||||
else:
|
||||
tenant_ids = UserModel.get_user_tenants(user_id)
|
||||
tenants = [TenantModel.get_by_id(tid) for tid in tenant_ids if TenantModel.get_by_id(tid)]
|
||||
|
||||
server_url = SettingsModel.get_server_url()
|
||||
return render_template('tenant_selection.html', tenants=tenants, server_url=server_url)
|
||||
|
||||
@main_bp.route('/settings', methods=['GET', 'POST'])
|
||||
@settings_access_required
|
||||
def settings():
|
||||
"""Global settings page"""
|
||||
"""Global settings page - admin only"""
|
||||
if request.method == 'POST':
|
||||
server_url = request.form.get('server_url', '').strip()
|
||||
if server_url:
|
||||
|
||||
@@ -1,10 +1,12 @@
|
||||
from flask import render_template, request, redirect, flash, jsonify, Response, current_app
|
||||
from flask import render_template, request, redirect, flash, jsonify, Response, current_app, session
|
||||
from . import tenant_bp
|
||||
from app.models import TenantModel, ProductModel, ARFieldModel, SettingsModel
|
||||
from app.models import TenantModel, ProductModel, ARFieldModel, SettingsModel, UserModel
|
||||
from app.services import AuthService, ProductService, BarcodeService
|
||||
from app.decorators.auth import tenant_access_required
|
||||
import os
|
||||
|
||||
@tenant_bp.route('/')
|
||||
@tenant_access_required
|
||||
def index(tenant_id):
|
||||
"""Tenant home page - product listing"""
|
||||
tenant = TenantModel.get_or_create(tenant_id)
|
||||
@@ -17,13 +19,19 @@ def index(tenant_id):
|
||||
return render_template('index.html', tenant=tenant, products=products, custom_fields=custom_fields)
|
||||
|
||||
@tenant_bp.route('/delete', methods=['POST'])
|
||||
@tenant_access_required
|
||||
def delete_tenant(tenant_id):
|
||||
"""Delete a tenant and all its data"""
|
||||
# Also remove the user's association with this tenant
|
||||
user_id = session['user']['id']
|
||||
UserModel.remove_tenant(user_id, tenant_id)
|
||||
|
||||
TenantModel.delete(tenant_id)
|
||||
flash(f'Tenant "{tenant_id}" has been deleted successfully.', 'success')
|
||||
return redirect('/')
|
||||
|
||||
@tenant_bp.route('/settings')
|
||||
@tenant_access_required
|
||||
def settings(tenant_id):
|
||||
"""Tenant settings page"""
|
||||
tenant = TenantModel.get_or_create(tenant_id)
|
||||
@@ -36,6 +44,7 @@ def settings(tenant_id):
|
||||
return render_template('tenant_settings.html', tenant=tenant, custom_fields=custom_fields, server_url=server_url)
|
||||
|
||||
@tenant_bp.route('/settings/credentials', methods=['POST'])
|
||||
@tenant_access_required
|
||||
def update_credentials(tenant_id):
|
||||
"""Update tenant credentials"""
|
||||
tenant = TenantModel.get_by_id(tenant_id)
|
||||
@@ -54,6 +63,7 @@ def update_credentials(tenant_id):
|
||||
return redirect(f'/{tenant_id}/settings')
|
||||
|
||||
@tenant_bp.route('/settings/barcode', methods=['POST'])
|
||||
@tenant_access_required
|
||||
def update_barcode_type(tenant_id):
|
||||
"""Update tenant barcode type"""
|
||||
tenant = TenantModel.get_by_id(tenant_id)
|
||||
|
||||
@@ -1,4 +1,8 @@
|
||||
import os
|
||||
from dotenv import load_dotenv
|
||||
|
||||
# Load environment variables from .env file
|
||||
load_dotenv()
|
||||
|
||||
class Config:
|
||||
"""Base configuration"""
|
||||
@@ -18,6 +22,23 @@ class Config:
|
||||
ALLOWED_EXTENSIONS = {'png', 'jpg', 'jpeg', 'gif'}
|
||||
MAX_CONTENT_LENGTH = 16 * 1024 * 1024 # 16MB max file size
|
||||
|
||||
# Session configuration
|
||||
SESSION_TYPE = 'filesystem'
|
||||
SESSION_PERMANENT = False
|
||||
SESSION_USE_SIGNER = True
|
||||
SESSION_FILE_DIR = os.path.join(DATA_FOLDER, 'flask_session')
|
||||
|
||||
# Entra ID (Azure AD) Configuration
|
||||
AZURE_CLIENT_ID = os.environ.get('AZURE_CLIENT_ID')
|
||||
AZURE_CLIENT_SECRET = os.environ.get('AZURE_CLIENT_SECRET')
|
||||
AZURE_TENANT_ID = os.environ.get('AZURE_TENANT_ID')
|
||||
REDIRECT_URI = os.environ.get('REDIRECT_URI', 'http://localhost:5555/auth/callback')
|
||||
AUTHORITY = os.environ.get('AUTHORITY', 'https://login.microsoftonline.com/')
|
||||
SCOPE = os.environ.get('SCOPE', 'User.Read').split()
|
||||
|
||||
# Admin configuration
|
||||
DEFAULT_ADMIN_EMAIL = os.environ.get('DEFAULT_ADMIN_EMAIL', '')
|
||||
|
||||
class DevelopmentConfig(Config):
|
||||
"""Development configuration"""
|
||||
DEBUG = True
|
||||
|
||||
91
src/app/decorators/auth.py
Normal file
91
src/app/decorators/auth.py
Normal file
@@ -0,0 +1,91 @@
|
||||
"""Authentication and authorization decorators"""
|
||||
from functools import wraps
|
||||
from flask import session, redirect, url_for, request, abort, current_app
|
||||
from app.models.user import UserModel
|
||||
|
||||
|
||||
def login_required(f):
|
||||
"""
|
||||
Decorator to require user authentication.
|
||||
Redirects to login if user is not authenticated.
|
||||
"""
|
||||
@wraps(f)
|
||||
def decorated_function(*args, **kwargs):
|
||||
if 'user' not in session:
|
||||
# Store the requested URL to redirect back after login
|
||||
session['next'] = request.url
|
||||
return redirect(url_for('auth.login'))
|
||||
return f(*args, **kwargs)
|
||||
return decorated_function
|
||||
|
||||
|
||||
def admin_required(f):
|
||||
"""
|
||||
Decorator to require admin role.
|
||||
Returns 403 if user is not an admin.
|
||||
"""
|
||||
@wraps(f)
|
||||
def decorated_function(*args, **kwargs):
|
||||
if 'user' not in session:
|
||||
session['next'] = request.url
|
||||
return redirect(url_for('auth.login'))
|
||||
|
||||
user_id = session['user'].get('id')
|
||||
if not UserModel.is_admin(user_id):
|
||||
current_app.logger.warning(f"User {user_id} attempted to access admin route without permission")
|
||||
abort(403, description="Admin access required")
|
||||
|
||||
return f(*args, **kwargs)
|
||||
return decorated_function
|
||||
|
||||
|
||||
def tenant_access_required(f):
|
||||
"""
|
||||
Decorator to require access to the specific tenant in the URL.
|
||||
User must either be an admin or have explicit access to the tenant.
|
||||
Admins can see server settings, regular users cannot.
|
||||
"""
|
||||
@wraps(f)
|
||||
def decorated_function(*args, **kwargs):
|
||||
if 'user' not in session:
|
||||
session['next'] = request.url
|
||||
return redirect(url_for('auth.login'))
|
||||
|
||||
user_id = session['user'].get('id')
|
||||
tenant_id = kwargs.get('tenant_id')
|
||||
|
||||
if not tenant_id:
|
||||
# No tenant in URL, allow access (e.g., global settings for admins only)
|
||||
return f(*args, **kwargs)
|
||||
|
||||
# Check if user has access to this tenant
|
||||
if not UserModel.has_access_to_tenant(user_id, tenant_id):
|
||||
current_app.logger.warning(
|
||||
f"User {user_id} attempted to access tenant {tenant_id} without permission"
|
||||
)
|
||||
abort(403, description="You don't have access to this tenant")
|
||||
|
||||
return f(*args, **kwargs)
|
||||
return decorated_function
|
||||
|
||||
|
||||
def settings_access_required(f):
|
||||
"""
|
||||
Decorator for server settings pages - admin only.
|
||||
Regular users cannot see server settings.
|
||||
"""
|
||||
@wraps(f)
|
||||
def decorated_function(*args, **kwargs):
|
||||
if 'user' not in session:
|
||||
session['next'] = request.url
|
||||
return redirect(url_for('auth.login'))
|
||||
|
||||
user_id = session['user'].get('id')
|
||||
if not UserModel.is_admin(user_id):
|
||||
current_app.logger.warning(
|
||||
f"User {user_id} attempted to access server settings without admin permission"
|
||||
)
|
||||
abort(403, description="Admin access required to view server settings")
|
||||
|
||||
return f(*args, **kwargs)
|
||||
return decorated_function
|
||||
@@ -2,5 +2,6 @@ from .tenant import TenantModel
|
||||
from .product import ProductModel
|
||||
from .ar_field import ARFieldModel
|
||||
from .settings import SettingsModel
|
||||
from .user import UserModel
|
||||
|
||||
__all__ = ['TenantModel', 'ProductModel', 'ARFieldModel', 'SettingsModel']
|
||||
__all__ = ['TenantModel', 'ProductModel', 'ARFieldModel', 'SettingsModel', 'UserModel']
|
||||
|
||||
253
src/app/models/user.py
Normal file
253
src/app/models/user.py
Normal file
@@ -0,0 +1,253 @@
|
||||
"""User model for authentication and authorization"""
|
||||
import sqlite3
|
||||
from datetime import datetime
|
||||
from typing import Optional, List, Dict
|
||||
from .base import get_db
|
||||
|
||||
|
||||
class UserModel:
|
||||
"""Model for managing users and their roles"""
|
||||
|
||||
ROLE_USER = 'user'
|
||||
ROLE_ADMIN = 'admin'
|
||||
|
||||
@staticmethod
|
||||
def create_table():
|
||||
"""Create the users and user_tenants tables if they don't exist"""
|
||||
with get_db() as conn:
|
||||
cursor = conn.cursor()
|
||||
|
||||
# Users table
|
||||
cursor.execute('''
|
||||
CREATE TABLE IF NOT EXISTS users (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
email TEXT UNIQUE NOT NULL,
|
||||
azure_oid TEXT UNIQUE,
|
||||
name TEXT,
|
||||
role TEXT NOT NULL DEFAULT 'user',
|
||||
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
||||
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
|
||||
)
|
||||
''')
|
||||
|
||||
# User-Tenant association table (users can own multiple tenants)
|
||||
cursor.execute('''
|
||||
CREATE TABLE IF NOT EXISTS user_tenants (
|
||||
user_id INTEGER NOT NULL,
|
||||
tenant_id TEXT NOT NULL,
|
||||
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
||||
PRIMARY KEY (user_id, tenant_id),
|
||||
FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE,
|
||||
FOREIGN KEY (tenant_id) REFERENCES tenants(id) ON DELETE CASCADE
|
||||
)
|
||||
''')
|
||||
|
||||
conn.commit()
|
||||
|
||||
@staticmethod
|
||||
def get_by_email(email: str) -> Optional[Dict]:
|
||||
"""Get user by email address"""
|
||||
with get_db() as conn:
|
||||
cursor = conn.cursor()
|
||||
cursor.execute(
|
||||
'SELECT id, email, azure_oid, name, role, created_at, updated_at FROM users WHERE email = ?',
|
||||
(email,)
|
||||
)
|
||||
row = cursor.fetchone()
|
||||
if row:
|
||||
return {
|
||||
'id': row[0],
|
||||
'email': row[1],
|
||||
'azure_oid': row[2],
|
||||
'name': row[3],
|
||||
'role': row[4],
|
||||
'created_at': row[5],
|
||||
'updated_at': row[6]
|
||||
}
|
||||
return None
|
||||
|
||||
@staticmethod
|
||||
def get_by_id(user_id: int) -> Optional[Dict]:
|
||||
"""Get user by ID"""
|
||||
with get_db() as conn:
|
||||
cursor = conn.cursor()
|
||||
cursor.execute(
|
||||
'SELECT id, email, azure_oid, name, role, created_at, updated_at FROM users WHERE id = ?',
|
||||
(user_id,)
|
||||
)
|
||||
row = cursor.fetchone()
|
||||
if row:
|
||||
return {
|
||||
'id': row[0],
|
||||
'email': row[1],
|
||||
'azure_oid': row[2],
|
||||
'name': row[3],
|
||||
'role': row[4],
|
||||
'created_at': row[5],
|
||||
'updated_at': row[6]
|
||||
}
|
||||
return None
|
||||
|
||||
@staticmethod
|
||||
def get_by_azure_oid(azure_oid: str) -> Optional[Dict]:
|
||||
"""Get user by Azure Object ID"""
|
||||
with get_db() as conn:
|
||||
cursor = conn.cursor()
|
||||
cursor.execute(
|
||||
'SELECT id, email, azure_oid, name, role, created_at, updated_at FROM users WHERE azure_oid = ?',
|
||||
(azure_oid,)
|
||||
)
|
||||
row = cursor.fetchone()
|
||||
if row:
|
||||
return {
|
||||
'id': row[0],
|
||||
'email': row[1],
|
||||
'azure_oid': row[2],
|
||||
'name': row[3],
|
||||
'role': row[4],
|
||||
'created_at': row[5],
|
||||
'updated_at': row[6]
|
||||
}
|
||||
return None
|
||||
|
||||
@staticmethod
|
||||
def create(email: str, name: str = None, azure_oid: str = None, role: str = ROLE_USER) -> Dict:
|
||||
"""Create a new user"""
|
||||
with get_db() as conn:
|
||||
cursor = conn.cursor()
|
||||
cursor.execute(
|
||||
'''INSERT INTO users (email, azure_oid, name, role)
|
||||
VALUES (?, ?, ?, ?)''',
|
||||
(email, azure_oid, name, role)
|
||||
)
|
||||
conn.commit()
|
||||
user_id = cursor.lastrowid
|
||||
return UserModel.get_by_id(user_id)
|
||||
|
||||
@staticmethod
|
||||
def update(user_id: int, **kwargs) -> Optional[Dict]:
|
||||
"""Update user information"""
|
||||
allowed_fields = {'email', 'azure_oid', 'name', 'role'}
|
||||
update_fields = {k: v for k, v in kwargs.items() if k in allowed_fields}
|
||||
|
||||
if not update_fields:
|
||||
return UserModel.get_by_id(user_id)
|
||||
|
||||
update_fields['updated_at'] = datetime.now().isoformat()
|
||||
|
||||
set_clause = ', '.join([f'{k} = ?' for k in update_fields.keys()])
|
||||
values = list(update_fields.values())
|
||||
values.append(user_id)
|
||||
|
||||
with get_db() as conn:
|
||||
cursor = conn.cursor()
|
||||
cursor.execute(
|
||||
f'UPDATE users SET {set_clause} WHERE id = ?',
|
||||
values
|
||||
)
|
||||
conn.commit()
|
||||
return UserModel.get_by_id(user_id)
|
||||
|
||||
@staticmethod
|
||||
def get_or_create_from_azure(email: str, name: str = None, azure_oid: str = None, default_admin_email: str = None) -> Dict:
|
||||
"""Get existing user or create new one from Azure AD login"""
|
||||
# Try to find by Azure OID first
|
||||
if azure_oid:
|
||||
user = UserModel.get_by_azure_oid(azure_oid)
|
||||
if user:
|
||||
# Update email/name if changed
|
||||
if user['email'] != email or user['name'] != name:
|
||||
return UserModel.update(user['id'], email=email, name=name)
|
||||
return user
|
||||
|
||||
# Try to find by email
|
||||
user = UserModel.get_by_email(email)
|
||||
if user:
|
||||
# Update Azure OID if not set
|
||||
if azure_oid and not user['azure_oid']:
|
||||
return UserModel.update(user['id'], azure_oid=azure_oid, name=name)
|
||||
return user
|
||||
|
||||
# Create new user
|
||||
# Case-insensitive comparison for admin email
|
||||
role = UserModel.ROLE_ADMIN if (default_admin_email and email.lower() == default_admin_email.lower()) else UserModel.ROLE_USER
|
||||
return UserModel.create(email=email, name=name, azure_oid=azure_oid, role=role)
|
||||
|
||||
@staticmethod
|
||||
def add_tenant(user_id: int, tenant_id: str):
|
||||
"""Associate a tenant with a user"""
|
||||
with get_db() as conn:
|
||||
cursor = conn.cursor()
|
||||
try:
|
||||
cursor.execute(
|
||||
'INSERT INTO user_tenants (user_id, tenant_id) VALUES (?, ?)',
|
||||
(user_id, tenant_id)
|
||||
)
|
||||
conn.commit()
|
||||
except sqlite3.IntegrityError:
|
||||
# Already exists, that's fine
|
||||
pass
|
||||
|
||||
@staticmethod
|
||||
def remove_tenant(user_id: int, tenant_id: str):
|
||||
"""Remove tenant association from user"""
|
||||
with get_db() as conn:
|
||||
cursor = conn.cursor()
|
||||
cursor.execute(
|
||||
'DELETE FROM user_tenants WHERE user_id = ? AND tenant_id = ?',
|
||||
(user_id, tenant_id)
|
||||
)
|
||||
conn.commit()
|
||||
|
||||
@staticmethod
|
||||
def get_user_tenants(user_id: int) -> List[str]:
|
||||
"""Get all tenant IDs associated with a user"""
|
||||
with get_db() as conn:
|
||||
cursor = conn.cursor()
|
||||
cursor.execute(
|
||||
'SELECT tenant_id FROM user_tenants WHERE user_id = ?',
|
||||
(user_id,)
|
||||
)
|
||||
return [row[0] for row in cursor.fetchall()]
|
||||
|
||||
@staticmethod
|
||||
def has_access_to_tenant(user_id: int, tenant_id: str) -> bool:
|
||||
"""Check if user has access to a specific tenant"""
|
||||
# Admin has access to everything
|
||||
user = UserModel.get_by_id(user_id)
|
||||
if user and user['role'] == UserModel.ROLE_ADMIN:
|
||||
return True
|
||||
|
||||
# Check if user owns the tenant
|
||||
with get_db() as conn:
|
||||
cursor = conn.cursor()
|
||||
cursor.execute(
|
||||
'SELECT 1 FROM user_tenants WHERE user_id = ? AND tenant_id = ?',
|
||||
(user_id, tenant_id)
|
||||
)
|
||||
return cursor.fetchone() is not None
|
||||
|
||||
@staticmethod
|
||||
def is_admin(user_id: int) -> bool:
|
||||
"""Check if user has admin role"""
|
||||
user = UserModel.get_by_id(user_id)
|
||||
return user and user['role'] == UserModel.ROLE_ADMIN
|
||||
|
||||
@staticmethod
|
||||
def get_all() -> List[Dict]:
|
||||
"""Get all users"""
|
||||
with get_db() as conn:
|
||||
cursor = conn.cursor()
|
||||
cursor.execute(
|
||||
'SELECT id, email, azure_oid, name, role, created_at, updated_at FROM users ORDER BY email'
|
||||
)
|
||||
rows = cursor.fetchall()
|
||||
return [{
|
||||
'id': row[0],
|
||||
'email': row[1],
|
||||
'azure_oid': row[2],
|
||||
'name': row[3],
|
||||
'role': row[4],
|
||||
'created_at': row[5],
|
||||
'updated_at': row[6]
|
||||
} for row in rows]
|
||||
129
src/app/services/msal_service.py
Normal file
129
src/app/services/msal_service.py
Normal file
@@ -0,0 +1,129 @@
|
||||
"""Microsoft Authentication Library (MSAL) service for Entra ID authentication"""
|
||||
import msal
|
||||
from flask import current_app, session, url_for
|
||||
from typing import Optional, Dict
|
||||
|
||||
|
||||
class MSALService:
|
||||
"""Service for handling Entra ID authentication using MSAL"""
|
||||
|
||||
@staticmethod
|
||||
def _get_msal_app():
|
||||
"""Get or create MSAL confidential client application"""
|
||||
authority = f"{current_app.config['AUTHORITY']}{current_app.config['AZURE_TENANT_ID']}"
|
||||
|
||||
return msal.ConfidentialClientApplication(
|
||||
current_app.config['AZURE_CLIENT_ID'],
|
||||
authority=authority,
|
||||
client_credential=current_app.config['AZURE_CLIENT_SECRET']
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def get_auth_url(state: str = None) -> str:
|
||||
"""
|
||||
Generate the authorization URL for Entra ID login
|
||||
|
||||
Args:
|
||||
state: Optional state parameter for CSRF protection
|
||||
|
||||
Returns:
|
||||
Authorization URL to redirect user to
|
||||
"""
|
||||
msal_app = MSALService._get_msal_app()
|
||||
|
||||
auth_url = msal_app.get_authorization_request_url(
|
||||
scopes=current_app.config['SCOPE'],
|
||||
state=state,
|
||||
redirect_uri=current_app.config['REDIRECT_URI']
|
||||
)
|
||||
|
||||
return auth_url
|
||||
|
||||
@staticmethod
|
||||
def acquire_token_by_auth_code(auth_code: str) -> Optional[Dict]:
|
||||
"""
|
||||
Exchange authorization code for access token
|
||||
|
||||
Args:
|
||||
auth_code: Authorization code received from callback
|
||||
|
||||
Returns:
|
||||
Token response dictionary or None if failed
|
||||
"""
|
||||
msal_app = MSALService._get_msal_app()
|
||||
|
||||
result = msal_app.acquire_token_by_authorization_code(
|
||||
auth_code,
|
||||
scopes=current_app.config['SCOPE'],
|
||||
redirect_uri=current_app.config['REDIRECT_URI']
|
||||
)
|
||||
|
||||
return result
|
||||
|
||||
@staticmethod
|
||||
def get_token_from_cache() -> Optional[Dict]:
|
||||
"""
|
||||
Try to get token from cache
|
||||
|
||||
Returns:
|
||||
Token response or None if no valid token in cache
|
||||
"""
|
||||
if 'user' not in session:
|
||||
return None
|
||||
|
||||
msal_app = MSALService._get_msal_app()
|
||||
accounts = msal_app.get_accounts()
|
||||
|
||||
if accounts:
|
||||
result = msal_app.acquire_token_silent(
|
||||
current_app.config['SCOPE'],
|
||||
account=accounts[0]
|
||||
)
|
||||
return result
|
||||
|
||||
return None
|
||||
|
||||
@staticmethod
|
||||
def remove_account():
|
||||
"""Remove account from MSAL cache"""
|
||||
msal_app = MSALService._get_msal_app()
|
||||
accounts = msal_app.get_accounts()
|
||||
|
||||
for account in accounts:
|
||||
msal_app.remove_account(account)
|
||||
|
||||
@staticmethod
|
||||
def get_user_info(token: str) -> Optional[Dict]:
|
||||
"""
|
||||
Get user information from Microsoft Graph API
|
||||
|
||||
Args:
|
||||
token: Access token
|
||||
|
||||
Returns:
|
||||
User info dictionary with email, name, etc.
|
||||
"""
|
||||
import requests
|
||||
|
||||
graph_endpoint = 'https://graph.microsoft.com/v1.0/me'
|
||||
|
||||
headers = {
|
||||
'Authorization': f'Bearer {token}',
|
||||
'Content-Type': 'application/json'
|
||||
}
|
||||
|
||||
try:
|
||||
response = requests.get(graph_endpoint, headers=headers)
|
||||
response.raise_for_status()
|
||||
user_data = response.json()
|
||||
|
||||
return {
|
||||
'email': user_data.get('mail') or user_data.get('userPrincipalName'),
|
||||
'name': user_data.get('displayName'),
|
||||
'azure_oid': user_data.get('id'),
|
||||
'given_name': user_data.get('givenName'),
|
||||
'surname': user_data.get('surname')
|
||||
}
|
||||
except requests.exceptions.RequestException as e:
|
||||
current_app.logger.error(f"Error fetching user info from Graph API: {e}")
|
||||
return None
|
||||
@@ -61,13 +61,36 @@
|
||||
<span class="navbar-toggler-icon"></span>
|
||||
</button>
|
||||
<div class="collapse navbar-collapse" id="navbarNav">
|
||||
<ul class="navbar-nav">
|
||||
<ul class="navbar-nav me-auto">
|
||||
<li class="nav-item">
|
||||
<a class="nav-link" href="/">Home</a>
|
||||
</li>
|
||||
{% if session.user and session.user.role == 'admin' %}
|
||||
<li class="nav-item">
|
||||
<a class="nav-link" href="/admin/">Admin</a>
|
||||
<a class="nav-link" href="/settings">Server Settings</a>
|
||||
</li>
|
||||
{% endif %}
|
||||
</ul>
|
||||
<ul class="navbar-nav ms-auto">
|
||||
{% if session.user %}
|
||||
<li class="nav-item dropdown">
|
||||
<a class="nav-link dropdown-toggle" href="#" id="userDropdown" role="button" data-bs-toggle="dropdown">
|
||||
<i class="bi bi-person-circle"></i> {{ session.user.name or session.user.email }}
|
||||
{% if session.user.role == 'admin' %}
|
||||
<span class="badge bg-danger ms-1">Admin</span>
|
||||
{% endif %}
|
||||
</a>
|
||||
<ul class="dropdown-menu dropdown-menu-end">
|
||||
<li><a class="dropdown-item disabled">{{ session.user.email }}</a></li>
|
||||
<li><hr class="dropdown-divider"></li>
|
||||
<li><a class="dropdown-item" href="/auth/logout"><i class="bi bi-box-arrow-right"></i> Logout</a></li>
|
||||
</ul>
|
||||
</li>
|
||||
{% else %}
|
||||
<li class="nav-item">
|
||||
<a class="nav-link" href="/auth/login"><i class="bi bi-box-arrow-in-right"></i> Login</a>
|
||||
</li>
|
||||
{% endif %}
|
||||
</ul>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
32
src/app/templates/unauthorized.html
Normal file
32
src/app/templates/unauthorized.html
Normal file
@@ -0,0 +1,32 @@
|
||||
{% extends "layout.html" %}
|
||||
|
||||
{% block title %}Access Denied - KCAP Demo Server{% endblock %}
|
||||
|
||||
{% block content %}
|
||||
<div class="row justify-content-center mt-5">
|
||||
<div class="col-md-6">
|
||||
<div class="card border-danger">
|
||||
<div class="card-header bg-danger text-white">
|
||||
<h4 class="mb-0"><i class="bi bi-shield-exclamation"></i> Access Denied</h4>
|
||||
</div>
|
||||
<div class="card-body text-center">
|
||||
<i class="bi bi-shield-exclamation" style="font-size: 4rem; color: #dc3545;"></i>
|
||||
<h5 class="mt-3">You don't have permission to access this resource</h5>
|
||||
<p class="text-muted">
|
||||
This page requires special permissions that your account doesn't have.
|
||||
Please contact an administrator if you believe this is an error.
|
||||
</p>
|
||||
<hr>
|
||||
<div class="d-grid gap-2 d-md-block">
|
||||
<a href="/" class="btn btn-primary"><i class="bi bi-house"></i> Go to Home</a>
|
||||
{% if session.user %}
|
||||
<a href="/auth/logout" class="btn btn-outline-secondary"><i class="bi bi-box-arrow-right"></i> Logout</a>
|
||||
{% else %}
|
||||
<a href="/auth/login" class="btn btn-outline-primary"><i class="bi bi-box-arrow-in-right"></i> Login</a>
|
||||
{% endif %}
|
||||
</div>
|
||||
</div>
|
||||
</div>
|
||||
</div>
|
||||
</div>
|
||||
{% endblock %}
|
||||
@@ -4,7 +4,9 @@ from app import create_app
|
||||
from app.models import TenantModel
|
||||
|
||||
# Create application
|
||||
app = create_app(os.environ.get('FLASK_ENV', 'development'))
|
||||
# Use FLASK_DEBUG instead of deprecated FLASK_ENV
|
||||
config_name = 'development' if os.environ.get('FLASK_DEBUG', '1') == '1' else 'production'
|
||||
app = create_app(config_name)
|
||||
|
||||
if __name__ == '__main__':
|
||||
with app.app_context():
|
||||
|
||||
Reference in New Issue
Block a user