Merge pull request #7 from mattintech/feature/cicd

Add GitHub issue templates
This commit is contained in:
2025-10-20 19:44:17 -04:00
committed by GitHub
20 changed files with 1111 additions and 22 deletions

24
.env.example Normal file
View File

@@ -0,0 +1,24 @@
# Flask Configuration
SECRET_KEY=your-secret-key-change-this-in-production
FLASK_DEBUG=1
# Entra ID (Azure AD) Configuration
# Get these values from your Azure Portal > App Registrations
AZURE_CLIENT_ID=your-client-id-here
AZURE_CLIENT_SECRET=your-client-secret-here
AZURE_TENANT_ID=your-tenant-id-here
# Redirect URI - Update this to match your deployment URL
# For local development: http://localhost:5555/auth/callback
# For production: https://yourdomain.com/auth/callback
REDIRECT_URI=http://localhost:5555/auth/callback
# Azure AD Authority URL (usually don't need to change this)
AUTHORITY=https://login.microsoftonline.com/
# Scopes for Microsoft Graph API (to get user info)
SCOPE=User.Read
# Application Settings
# Default admin email (this user will have admin role automatically)
DEFAULT_ADMIN_EMAIL=admin@yourdomain.com

127
.github/ISSUE_TEMPLATE/bug_report.yml vendored Normal file
View File

@@ -0,0 +1,127 @@
name: Bug Report
description: Report a bug or issue with the Flask AR API Demo Server
title: "[Bug]: "
labels: ["bug"]
body:
- type: markdown
attributes:
value: |
Thanks for taking the time to report a bug! Please fill out the information below to help us resolve the issue.
- type: dropdown
id: component
attributes:
label: Component
description: Which component is affected?
options:
- API Endpoints (login, arcontentfields, arinfo)
- Admin Interface
- Barcode Generation
- Image Serving
- Authentication (Entra ID)
- Database/Models
- Other
validations:
required: true
- type: textarea
id: description
attributes:
label: Bug Description
description: A clear and concise description of what the bug is.
placeholder: Describe the bug...
validations:
required: true
- type: textarea
id: reproduction
attributes:
label: Steps to Reproduce
description: Steps to reproduce the behavior
placeholder: |
1. Navigate to '...'
2. Click on '...'
3. Make API call to '...'
4. See error
validations:
required: true
- type: textarea
id: expected
attributes:
label: Expected Behavior
description: What did you expect to happen?
placeholder: Describe what should happen...
validations:
required: true
- type: textarea
id: actual
attributes:
label: Actual Behavior
description: What actually happened?
placeholder: Describe what actually happened...
validations:
required: true
- type: textarea
id: logs
attributes:
label: Error Logs/Screenshots
description: If applicable, add error logs, stack traces, or screenshots
placeholder: Paste logs or drag and drop screenshots here...
validations:
required: false
- type: dropdown
id: environment
attributes:
label: Environment
description: Where are you running the application?
options:
- Local Development
- Docker
- Cloud Deployment
- Other
validations:
required: true
- type: input
id: python-version
attributes:
label: Python Version
description: What version of Python are you using?
placeholder: e.g., 3.9.0, 3.10.2, 3.11.5
validations:
required: false
- type: input
id: flask-version
attributes:
label: Flask Version
description: What version of Flask are you using?
placeholder: e.g., 2.2.5
validations:
required: false
- type: textarea
id: additional
attributes:
label: Additional Context
description: Add any other context about the problem here
placeholder: Any additional information...
validations:
required: false
- type: checkboxes
id: checklist
attributes:
label: Pre-submission Checklist
description: Please confirm the following
options:
- label: I have searched existing issues to ensure this bug hasn't been reported
required: true
- label: I have included all relevant information above
required: true
- label: I can reproduce this bug consistently
required: false

View File

@@ -0,0 +1,155 @@
name: Feature Request
description: Suggest a new feature or enhancement for the Flask AR API Demo Server
title: "[Feature]: "
labels: ["enhancement"]
body:
- type: markdown
attributes:
value: |
Thanks for suggesting a new feature! Please provide as much detail as possible to help us understand your request.
- type: dropdown
id: component
attributes:
label: Component
description: Which component would this feature affect?
options:
- API Endpoints
- Admin Interface
- Barcode Generation
- Image Management
- Authentication/Authorization
- Database/Models
- Product Management
- Tenant Management
- Documentation
- Other
validations:
required: true
- type: dropdown
id: feature-type
attributes:
label: Feature Type
description: What type of feature is this?
options:
- New Functionality
- Enhancement to Existing Feature
- UI/UX Improvement
- Performance Improvement
- Developer Experience
- Documentation
- Testing
- Security
- Other
validations:
required: true
- type: textarea
id: problem
attributes:
label: Problem Statement
description: Is your feature request related to a problem? Please describe.
placeholder: I'm frustrated when... / There's no way to... / It would be helpful if...
validations:
required: true
- type: textarea
id: solution
attributes:
label: Proposed Solution
description: Describe the solution you'd like to see
placeholder: I would like to be able to... / The system should... / Add a new endpoint that...
validations:
required: true
- type: textarea
id: alternatives
attributes:
label: Alternatives Considered
description: Have you considered any alternative solutions or workarounds?
placeholder: I've thought about... / Another approach could be... / Currently using...
validations:
required: false
- type: textarea
id: use-case
attributes:
label: Use Case
description: Describe your use case and how this feature would benefit you or others
placeholder: |
This feature would allow me to...
It would be useful for...
The expected workflow would be...
validations:
required: true
- type: textarea
id: implementation
attributes:
label: Implementation Ideas
description: If you have ideas about how to implement this feature, please share them
placeholder: |
This could be implemented by...
Possible technical approach...
Required changes...
validations:
required: false
- type: dropdown
id: priority
attributes:
label: Priority
description: How important is this feature to you?
options:
- Nice to have
- Would be useful
- Important
- Critical
validations:
required: true
- type: dropdown
id: breaking
attributes:
label: Breaking Change
description: Would this feature require breaking changes to the API or existing functionality?
options:
- "No"
- "Yes"
- "Unsure"
validations:
required: true
- type: textarea
id: mockups
attributes:
label: Mockups/Examples
description: If applicable, add mockups, screenshots, or examples from other systems
placeholder: Drag and drop images or paste links here...
validations:
required: false
- type: textarea
id: additional
attributes:
label: Additional Context
description: Add any other context, links, or references about the feature request
placeholder: Related to... / Similar to... / Reference documentation...
validations:
required: false
- type: checkboxes
id: checklist
attributes:
label: Pre-submission Checklist
description: Please confirm the following
options:
- label: I have searched existing issues to ensure this feature hasn't been requested
required: true
- label: I have clearly described the feature and its benefits
required: true
- label: I am willing to help test this feature if implemented
required: false
- label: I am willing to contribute to the implementation
required: false

7
.gitignore vendored
View File

@@ -6,3 +6,10 @@ __pycache__/
*.pyo
*.pyd
.claude/
# Environment variables
.env
.env.local
# Flask session files
flask_session/

View File

@@ -1,5 +1,9 @@
Flask==2.0.1
Werkzeug==2.0.3
Flask==2.2.5
Werkzeug==2.2.3
python-barcode==0.14.0
qrcode==7.4.2
Pillow==9.5.0
msal==1.24.0
python-dotenv==1.0.0
Flask-Session==0.5.0
requests==2.31.0

View File

@@ -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)

View File

@@ -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)

View File

@@ -0,0 +1,6 @@
"""Authentication blueprint"""
from flask import Blueprint
auth_bp = Blueprint('auth', __name__, url_prefix='/auth')
from . import routes

View 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

View File

@@ -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"""
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:

View File

@@ -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)

View File

@@ -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

View 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

View File

@@ -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
View 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]

View 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

View File

@@ -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>

View File

@@ -119,10 +119,10 @@
<button type="button" class="btn btn-sm btn-outline-primary" data-bs-toggle="modal" data-bs-target="#qrModal-{{ tenant.id }}">
<i class="bi bi-qr-code"></i> QR
</button>
<form method="POST" action="/tenant/{{ tenant.id }}/delete" class="d-inline" onsubmit="return confirmDelete('{{ tenant.name }}')">
<button type="submit" class="btn btn-danger btn-sm">
<button type="button" class="btn btn-danger btn-sm" data-bs-toggle="modal" data-bs-target="#deleteTenantModal" data-tenant-id="{{ tenant.id }}" data-tenant-name="{{ tenant.name }}">
<i class="bi bi-trash"></i> Delete
</button>
<form id="delete-tenant-form-{{ tenant.id }}" method="POST" action="/{{ tenant.id }}/delete" style="display: none;">
</form>
</div>
</div>
@@ -198,6 +198,35 @@
</div>
{% endfor %}
<!-- Delete Tenant Confirmation Modal -->
<div class="modal fade" id="deleteTenantModal" tabindex="-1" aria-labelledby="deleteTenantModalLabel" aria-hidden="true">
<div class="modal-dialog modal-dialog-centered">
<div class="modal-content">
<div class="modal-header bg-danger text-white">
<h5 class="modal-title" id="deleteTenantModalLabel">
<i class="bi bi-exclamation-triangle-fill me-2"></i>Confirm Tenant Deletion
</h5>
<button type="button" class="btn-close btn-close-white" data-bs-dismiss="modal" aria-label="Close"></button>
</div>
<div class="modal-body">
<p class="mb-3">Are you sure you want to delete the tenant <strong id="delete-tenant-name"></strong>?</p>
<div class="alert alert-danger mb-0">
<i class="bi bi-exclamation-triangle-fill me-2"></i>
<strong>Warning:</strong> This will permanently delete all products and data associated with this tenant. This action cannot be undone.
</div>
</div>
<div class="modal-footer">
<button type="button" class="btn btn-secondary" data-bs-dismiss="modal">
<i class="bi bi-x-circle me-1"></i>Cancel
</button>
<button type="button" class="btn btn-danger" id="confirm-delete-tenant-btn">
<i class="bi bi-trash me-1"></i>Delete Tenant
</button>
</div>
</div>
</div>
</div>
<script src="https://cdn.jsdelivr.net/npm/bootstrap@5.3.0/dist/js/bootstrap.bundle.min.js"></script>
<script>
const reservedIds = ['admin', 'api', 'login', 'logout', 'arcontentfields', 'arinfo',
@@ -259,9 +288,27 @@
searchInput.focus();
});
// Delete confirmation
function confirmDelete(tenantName) {
return confirm(`Are you sure you want to delete the tenant "${tenantName}"?\n\nThis will permanently delete all products and data associated with this tenant.`);
// Delete tenant confirmation modal functionality
const deleteTenantModal = document.getElementById('deleteTenantModal');
if (deleteTenantModal) {
deleteTenantModal.addEventListener('show.bs.modal', function (event) {
// Button that triggered the modal
const button = event.relatedTarget;
// Extract tenant info from data attributes
const tenantId = button.getAttribute('data-tenant-id');
const tenantName = button.getAttribute('data-tenant-name');
// Update modal content
const modalTenantName = document.getElementById('delete-tenant-name');
modalTenantName.textContent = `"${tenantName}" (ID: ${tenantId})`;
// Setup the confirm button action
const confirmDeleteBtn = document.getElementById('confirm-delete-tenant-btn');
confirmDeleteBtn.onclick = function() {
document.getElementById('delete-tenant-form-' + tenantId).submit();
};
});
}
</script>
</body>

View 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 %}

View File

@@ -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():