import pymysql import bcrypt import jwt from datetime import datetime, timedelta from functools import wraps from flask import request, jsonify import logging import secrets logger = logging.getLogger(__name__) class AuthHandler: VALID_ROLES = {'admin', 'business_admin', 'manager', 'user', 'agent'} ADMIN_ROLES = {'admin', 'business_admin'} def __init__(self, config): self.config = config self.db_config = { 'host': config.get('DB_HOST'), 'port': int(config.get('DB_PORT', 3306)), 'user': config.get('DB_USER'), 'password': config.get('DB_PASSWORD'), 'database': config.get('DB_NAME'), 'cursorclass': pymysql.cursors.DictCursor } # JWT secret key - should be in config self.jwt_secret = config.get('SECRET_KEY', 'your-secret-key-here-change-in-production') self.jwt_algorithm = 'HS256' self.jwt_expiration_days = 7 try: self.ensure_embed_tables() self.ensure_rbac_tables() except Exception as e: logger.warning("Could not ensure auth extension tables at startup (DB may be unreachable): %s", e) def get_connection(self): """Get database connection""" return pymysql.connect(**self.db_config) @staticmethod def _quote_identifier(identifier): return f"`{identifier.replace('`', '``')}`" def _table_exists(self, cursor, schema, table_name): cursor.execute( """SELECT 1 FROM information_schema.tables WHERE table_schema = %s AND table_name = %s LIMIT 1""", (schema, table_name) ) return cursor.fetchone() is not None def _find_template_table(self, cursor, schema, suffix): preferred = f"7987_{suffix}" cursor.execute( """SELECT table_name AS name FROM information_schema.tables WHERE table_schema = %s AND table_name = %s LIMIT 1""", (schema, preferred) ) row = cursor.fetchone() if row: return row['name'] cursor.execute( """SELECT table_name AS name FROM information_schema.tables WHERE table_schema = %s AND table_name LIKE %s ORDER BY table_name LIMIT 1""", (schema, f"%_{suffix}") ) row = cursor.fetchone() return row['name'] if row else None def hash_password(self, password): """Hash a password using bcrypt""" salt = bcrypt.gensalt() return bcrypt.hashpw(password.encode('utf-8'), salt).decode('utf-8') @staticmethod def _looks_like_bcrypt_hash(password_hash): if not password_hash or not isinstance(password_hash, str): return False if not password_hash.startswith(("$2a$", "$2b$", "$2y$")): return False # Typical bcrypt hash length is 60 characters. return len(password_hash) >= 50 def verify_password(self, password, password_hash): """Verify a password against its hash""" if not self._looks_like_bcrypt_hash(password_hash): return False @classmethod def normalize_role(cls, role): role = str(role or 'user').strip().lower() if role == 'business-admin': role = 'business_admin' return role def _ensure_role_column_support(self, cursor): """Expand legacy ENUM('admin','user','viewer') role columns for RBAC roles.""" for table in ('business_users', 'user_business_access'): cursor.execute( """ SELECT COLUMN_TYPE FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = %s AND COLUMN_NAME = 'role' """, (table,), ) row = cursor.fetchone() if not row: continue column_type = str((row or {}).get('COLUMN_TYPE') or '').lower() if 'enum' in column_type or 'viewer' in column_type: cursor.execute( f"UPDATE `{table}` SET role = 'user' WHERE role = 'viewer'" ) cursor.execute( f"ALTER TABLE `{table}` MODIFY COLUMN role VARCHAR(30) NOT NULL DEFAULT 'user'" ) logger.info("Expanded role column on %s for RBAC roles", table) def ensure_rbac_tables(self): """Create additive role/scope tables without changing existing auth tables.""" conn = self.get_connection() try: cursor = conn.cursor() self._ensure_role_column_support(cursor) cursor.execute(""" CREATE TABLE IF NOT EXISTS permissions ( code VARCHAR(100) PRIMARY KEY, description VARCHAR(255) DEFAULT NULL, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci """) cursor.execute(""" CREATE TABLE IF NOT EXISTS role_permissions ( role VARCHAR(30) NOT NULL, permission_code VARCHAR(100) NOT NULL, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, PRIMARY KEY (role, permission_code), INDEX idx_permission_code (permission_code) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci """) cursor.execute(""" CREATE TABLE IF NOT EXISTS user_agent_mapping ( id INT AUTO_INCREMENT PRIMARY KEY, user_id INT NOT NULL, bid VARCHAR(20) NOT NULL, agent_name VARCHAR(255) DEFAULT NULL, agent_phone VARCHAR(100) DEFAULT NULL, agent_extension VARCHAR(100) DEFAULT NULL, is_primary TINYINT(1) DEFAULT 0, is_active TINYINT(1) DEFAULT 1, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, UNIQUE KEY uq_user_bid_agent (user_id, bid, agent_name, agent_phone, agent_extension), INDEX idx_user_bid (user_id, bid), INDEX idx_bid_agent_name (bid, agent_name), INDEX idx_bid_agent_phone (bid, agent_phone), FOREIGN KEY (user_id) REFERENCES business_users(id) ON DELETE CASCADE ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci """) cursor.execute(""" CREATE TABLE IF NOT EXISTS user_group_mapping ( id INT AUTO_INCREMENT PRIMARY KEY, user_id INT NOT NULL, bid VARCHAR(20) NOT NULL, groupname VARCHAR(255) NOT NULL, is_active TINYINT(1) DEFAULT 1, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, UNIQUE KEY uq_user_bid_group (user_id, bid, groupname), INDEX idx_user_bid (user_id, bid), INDEX idx_bid_group (bid, groupname), FOREIGN KEY (user_id) REFERENCES business_users(id) ON DELETE CASCADE ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci """) cursor.execute(""" CREATE TABLE IF NOT EXISTS user_permission_overrides ( id INT AUTO_INCREMENT PRIMARY KEY, user_id INT NOT NULL, bid VARCHAR(20) NOT NULL, permission_code VARCHAR(100) NOT NULL, allowed TINYINT(1) NOT NULL, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, UNIQUE KEY uq_user_bid_permission (user_id, bid, permission_code), INDEX idx_user_bid (user_id, bid), FOREIGN KEY (user_id) REFERENCES business_users(id) ON DELETE CASCADE ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci """) default_permissions = { 'admin': [ 'users.view', 'users.create', 'users.update', 'users.disable', 'users.reset_password', 'roles.assign', 'agents.manage', 'calls.view_all', 'calls.view_detail', 'calls.view_transcript', 'calls.edit_transcript', 'calls.delete_transcript', 'calls.reprocess', 'leads.view_all', 'leads.update', 'analytics.view_all', 'analytics.leaderboard', 'exports.download', 'crm.manage', 'quality.manage', 'settings.manage', 'pipeline.manage', 'audit.view_business', ], 'business_admin': [ 'users.view', 'users.create', 'users.update', 'users.disable', 'users.reset_password', 'roles.assign', 'agents.manage', 'calls.view_all', 'calls.view_detail', 'calls.view_transcript', 'calls.edit_transcript', 'calls.delete_transcript', 'calls.reprocess', 'leads.view_all', 'leads.update', 'analytics.view_all', 'analytics.leaderboard', 'exports.download', 'crm.manage', 'quality.manage', 'settings.manage', 'pipeline.manage', 'audit.view_business', ], 'manager': [ 'users.view', 'calls.view_team', 'calls.view_detail', 'calls.view_transcript', 'leads.view_team', 'analytics.view_team', 'analytics.leaderboard_team', 'exports.team_download', ], 'user': [ 'dashboard.view', 'calls.view_all', 'calls.view_detail', 'calls.view_transcript', 'leads.view_all', 'analytics.view_all', ], 'agent': [ 'dashboard.view', 'calls.view_own', 'calls.view_detail', 'calls.view_transcript_own', 'leads.view_own', 'analytics.view_own', 'profile.manage_own', ], } descriptions = { code: code.replace('_', ' ').replace('.', ': ') for perms in default_permissions.values() for code in perms } for code, description in descriptions.items(): cursor.execute( "INSERT IGNORE INTO permissions (code, description) VALUES (%s, %s)", (code, description[:255]), ) for role, perms in default_permissions.items(): for code in perms: cursor.execute( "INSERT IGNORE INTO role_permissions (role, permission_code) VALUES (%s, %s)", (role, code), ) conn.commit() except Exception as e: conn.rollback() logger.error("Error ensuring RBAC tables: %s", e) raise finally: conn.close() def ensure_user_activity_log_audit_columns(self): """Add bid / business_name / action_code if missing (additive ALTER).""" conn = self.get_connection() try: cursor = conn.cursor() cursor.execute( """ SELECT COLUMN_NAME FROM information_schema.COLUMNS WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'user_activity_log' """ ) existing = {r["COLUMN_NAME"] for r in (cursor.fetchall() or [])} parts = [] if "bid" not in existing: parts.append("ADD COLUMN bid VARCHAR(64) DEFAULT NULL") if "business_name" not in existing: parts.append("ADD COLUMN business_name VARCHAR(255) DEFAULT NULL") if "action_code" not in existing: parts.append("ADD COLUMN action_code VARCHAR(32) DEFAULT NULL") if parts: cursor.execute("ALTER TABLE user_activity_log " + ", ".join(parts)) conn.commit() except Exception as e: logger.warning("Could not extend user_activity_log columns: %s", e) try: conn.rollback() except Exception: pass finally: conn.close() @staticmethod def _infer_action_code(activity_type): if not activity_type: return None t = str(activity_type).strip().lower() if t == "login": return "LOGIN" if t == "logout": return "LOGOUT" if t.startswith("delete") or t.startswith("revoke") or t == "delete": return "DELETE" if t.startswith("create") or t == "register": return "CREATE" if "change" in t or "update" in t or t.endswith("_update"): return "UPDATE" if "retry" in t or "reprocess" in t: return "RETRY" return None def log_activity( self, user_id, username, activity_type, description=None, ip_address=None, user_agent=None, bid=None, business_name=None, action_code=None, ): """Persist audit events (distinct from orchestration file logs).""" self.ensure_user_activity_log_audit_columns() if action_code is None: action_code = self._infer_action_code(activity_type) conn = self.get_connection() try: cursor = conn.cursor() cursor.execute( """INSERT INTO user_activity_log (user_id, username, activity_type, description, ip_address, user_agent, bid, business_name, action_code) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)""", ( user_id, username, activity_type, description, ip_address, user_agent, bid, business_name, action_code, ), ) conn.commit() except Exception as e: logger.error(f"Error logging activity: {str(e)}") finally: conn.close() def generate_jwt_token(self, user_id, username, email, businesses, is_master=False): """Generate a JWT token with user info and accessible businesses""" payload = { 'user_id': user_id, 'username': username, 'email': email, 'businesses': businesses, # List of {bid, name, role} 'is_master': is_master, 'exp': datetime.utcnow() + timedelta(days=self.jwt_expiration_days), 'iat': datetime.utcnow() } token = jwt.encode(payload, self.jwt_secret, algorithm=self.jwt_algorithm) return token def decode_jwt_token(self, token): """Decode and validate a JWT token""" try: payload = jwt.decode(token, self.jwt_secret, algorithms=[self.jwt_algorithm]) return payload except jwt.ExpiredSignatureError: return None except jwt.InvalidTokenError: return None def create_user(self, username, email, password, full_name=None, role='user', is_master=False): """Create a new user (no business assignment here)""" conn = self.get_connection() try: cursor = conn.cursor() role = self.normalize_role(role) if role not in self.VALID_ROLES: return {'error': 'Invalid role. Use admin, business_admin, manager, user, or agent.'}, 400 # Check if username or email already exists cursor.execute( "SELECT id FROM business_users WHERE username = %s OR email = %s", (username, email) ) if cursor.fetchone(): return {'error': 'Username or email already exists'}, 409 # Hash password password_hash = self.hash_password(password) # Insert user cursor.execute( """INSERT INTO business_users (username, email, password_hash, plain_password, full_name, role, is_master, is_active) VALUES (%s, %s, %s, %s, %s, %s, %s, TRUE)""", (username, email, password_hash, password, full_name, role, is_master) ) conn.commit() user_id = cursor.lastrowid return { 'id': user_id, 'username': username, 'email': email, 'role': role, 'is_master': is_master }, 201 except Exception as e: conn.rollback() logger.error(f"Error creating user: {str(e)}") return {'error': str(e)}, 500 finally: conn.close() def change_user_password(self, user_id, previous_password, new_password): """Change a user's password after verifying the existing password.""" conn = self.get_connection() try: cursor = conn.cursor() has_plain_password2 = False try: cursor.execute( """SELECT 1 FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'business_users' AND COLUMN_NAME = 'plain_password2' LIMIT 1""" ) has_plain_password2 = cursor.fetchone() is not None except Exception: has_plain_password2 = False cursor.execute( """SELECT id, username, password_hash, plain_password FROM business_users WHERE id = %s AND is_active = TRUE""", (user_id,) ) user = cursor.fetchone() if not user: return {'error': 'User not found'}, 404 password_ok = self.verify_password(previous_password, user.get('password_hash')) plain_password = user.get('plain_password') if not password_ok and plain_password and previous_password == plain_password: password_ok = True if not password_ok and user.get('password_hash') and previous_password == user.get('password_hash'): password_ok = True if not password_ok: return {'error': 'Previous password is incorrect'}, 401 if has_plain_password2: cursor.execute( """UPDATE business_users SET password_hash = %s, plain_password = %s, plain_password2 = %s WHERE id = %s""", (self.hash_password(new_password), new_password, new_password, user_id) ) else: cursor.execute( """UPDATE business_users SET password_hash = %s, plain_password = %s WHERE id = %s""", (self.hash_password(new_password), new_password, user_id) ) conn.commit() return {'message': 'Password changed successfully'}, 200 except Exception as e: conn.rollback() logger.error(f"Error changing user password: {str(e)}") return {'error': str(e)}, 500 finally: conn.close() def get_business_login_user(self, bid, user_id=None): """Return the primary login user linked to a business.""" conn = self.get_connection() try: cursor = conn.cursor() if user_id: cursor.execute( """SELECT bu.id, bu.username, bu.email, bu.plain_password FROM business_users bu JOIN user_business_access uba ON uba.user_id = bu.id WHERE uba.bid = %s AND bu.id = %s AND bu.is_active = TRUE LIMIT 1""", (bid, user_id), ) else: cursor.execute( """SELECT bu.id, bu.username, bu.email, bu.plain_password FROM business_users bu JOIN user_business_access uba ON uba.user_id = bu.id WHERE uba.bid = %s AND bu.is_active = TRUE AND bu.is_master = FALSE ORDER BY bu.id ASC LIMIT 1""", (bid,), ) return cursor.fetchone() finally: conn.close() def update_business_user_credentials(self, bid, username, email, password, user_id=None): """Master-admin update of a business login user's credentials.""" conn = self.get_connection() try: cursor = conn.cursor() target = self.get_business_login_user(bid, user_id=user_id) if not target: return {'error': 'Business login user not found'}, 404 target_id = target['id'] username = str(username or '').strip() email = str(email or '').strip().lower() password = str(password or '') if not username: return {'error': 'Username is required'}, 400 if not email: return {'error': 'Email is required'}, 400 if not password: return {'error': 'Password is required'}, 400 cursor.execute( """SELECT id FROM business_users WHERE (username = %s OR email = %s) AND id != %s LIMIT 1""", (username, email, target_id), ) if cursor.fetchone(): return {'error': 'Username or email already exists'}, 409 cursor.execute( """UPDATE business_users SET username = %s, email = %s, password_hash = %s, plain_password = %s WHERE id = %s""", (username, email, self.hash_password(password), password, target_id), ) conn.commit() return { 'business_user_id': target_id, 'credentials': { 'username': username, 'password': password, 'email': email, }, }, 200 except Exception as e: conn.rollback() logger.error(f"Error updating business credentials: {str(e)}") return {'error': str(e)}, 500 finally: conn.close() def assign_business_access(self, user_id, bid, role='user'): """Assign business access to a user""" conn = self.get_connection() try: cursor = conn.cursor() role = self.normalize_role(role) if role not in self.VALID_ROLES: return {'error': 'Invalid role. Use admin, business_admin, manager, user, or agent.'}, 400 # Check if business exists cursor.execute("SELECT bid FROM businesses WHERE bid = %s", (bid,)) if not cursor.fetchone(): return {'error': 'Business not found'}, 404 # Check if user exists cursor.execute("SELECT id FROM business_users WHERE id = %s", (user_id,)) if not cursor.fetchone(): return {'error': 'User not found'}, 404 # Insert access cursor.execute( """INSERT INTO user_business_access (user_id, bid, role) VALUES (%s, %s, %s) ON DUPLICATE KEY UPDATE role = %s""", (user_id, bid, role, role) ) conn.commit() return {'message': 'Business access granted', 'user_id': user_id, 'bid': bid, 'role': role}, 200 except Exception as e: conn.rollback() logger.error(f"Error assigning business access: {str(e)}") return {'error': str(e)}, 500 finally: conn.close() def get_user_businesses(self, user_id): """Get all businesses accessible to a user""" conn = self.get_connection() try: cursor = conn.cursor() cursor.execute( """SELECT b.bid, b.name, b.description, uba.role FROM user_business_access uba JOIN businesses b ON uba.bid = b.bid WHERE uba.user_id = %s AND b.is_active = TRUE ORDER BY b.name""", (user_id,) ) businesses = cursor.fetchall() for business in businesses: role = self.normalize_role(business.get('role')) permissions = self.get_permissions_for_user(user_id, business.get('bid'), role) scope = self.get_user_scope(user_id, business.get('bid'), role) business['role'] = role business['permissions'] = permissions business['scope'] = scope return businesses except Exception as e: logger.error(f"Error getting user businesses: {str(e)}") return [] finally: conn.close() def login(self, username, password, ip_address=None, user_agent=None): """Authenticate user and create JWT token""" conn = self.get_connection() try: cursor = conn.cursor() # Get user by username or email cursor.execute( """SELECT id, username, email, password_hash, plain_password, plain_password2, full_name, role, is_master, is_active FROM business_users WHERE (username = %s OR email = %s) AND is_active = TRUE""", (username, username) ) user = cursor.fetchone() if not user: return {'error': 'Invalid credentials'}, 401 # Verify password password_ok = self.verify_password(password, user.get('password_hash')) needs_upgrade = False if not password_ok: plain_password = user.get('plain_password') plain_password2 = user.get('plain_password2') if plain_password and password == plain_password: password_ok = True needs_upgrade = True elif plain_password2 and password == plain_password2: password_ok = True needs_upgrade = True elif user.get('password_hash') and password == user.get('password_hash'): # Handle legacy rows where password_hash stores plain text password_ok = True needs_upgrade = True if not password_ok: return {'error': 'Invalid credentials'}, 401 if needs_upgrade: new_hash = self.hash_password(password) cursor.execute( "UPDATE business_users SET password_hash = %s WHERE id = %s", (new_hash, user['id']) ) logger.warning( "Upgraded password hash for user %s due to legacy/invalid hash", user['username'] ) # Get user's accessible businesses businesses = self.get_user_businesses(user['id']) # For master users, they can access all businesses if user['is_master']: cursor.execute( """SELECT bid, name, description, 'admin' as role FROM businesses WHERE is_active = TRUE ORDER BY name""" ) businesses = cursor.fetchall() for business in businesses: business['permissions'] = ['*'] business['scope'] = {'type': 'all'} # Update last login cursor.execute( "UPDATE business_users SET last_login = NOW() WHERE id = %s", (user['id'],) ) conn.commit() # Generate JWT token token = self.generate_jwt_token( user_id=user['id'], username=user['username'], email=user['email'], businesses=businesses, is_master=user['is_master'] ) # Log login activity self.log_activity( user_id=user['id'], username=user['username'], activity_type='login', description=f"User {user['username']} logged in successfully", ip_address=ip_address, user_agent=user_agent, action_code='LOGIN', ) return { 'token': token, 'user': { 'id': user['id'], 'username': user['username'], 'email': user['email'], 'full_name': user['full_name'], 'role': self.normalize_role(user['role']), 'is_master': user['is_master'], 'businesses': businesses } }, 200 except Exception as e: conn.rollback() logger.error(f"Error during login: {str(e)}") return {'error': str(e)}, 500 finally: conn.close() def validate_token(self, token): """Validate a JWT token and return user info""" payload = self.decode_jwt_token(token) if not payload: return None return { 'id': payload.get('user_id'), 'username': payload.get('username'), 'email': payload.get('email'), 'businesses': payload.get('businesses', []), 'is_master': payload.get('is_master', False) } def logout(self, token): """Logout user (with JWT, we just rely on token expiration)""" # With JWT, logout is handled client-side by removing the token # We could implement a token blacklist here if needed return {'message': 'Logged out successfully'}, 200 def get_all_users(self): """Get all users (master admin only)""" conn = self.get_connection() try: cursor = conn.cursor() cursor.execute( """SELECT id, username, email, plain_password, full_name, role, is_master, is_active, created_at, last_login FROM business_users ORDER BY created_at DESC""" ) users = cursor.fetchall() # Get businesses for each user for user in users: user['businesses'] = self.get_user_businesses(user['id']) return users, 200 except Exception as e: logger.error(f"Error getting users: {str(e)}") return {'error': str(e)}, 500 finally: conn.close() def get_role_base_permissions(self, role='user'): """Return default permission codes for a role (no user overrides).""" role = self.normalize_role(role) conn = self.get_connection() try: cursor = conn.cursor() cursor.execute( "SELECT permission_code FROM role_permissions WHERE role = %s", (role,), ) return sorted({row['permission_code'] for row in (cursor.fetchall() or [])}) except Exception as e: logger.warning("Could not load base permissions for role=%s: %s", role, e) return [] finally: conn.close() def sync_permission_overrides(self, user_id, bid, role, desired_permissions): """Persist overrides so effective permissions match desired_permissions list.""" role = self.normalize_role(role) base = set(self.get_role_base_permissions(role)) desired = {str(code).strip() for code in (desired_permissions or []) if str(code).strip()} conn = self.get_connection() try: cursor = conn.cursor() cursor.execute( "DELETE FROM user_permission_overrides WHERE user_id = %s AND bid = %s", (user_id, str(bid)), ) for code in sorted(desired - base): cursor.execute( """INSERT INTO user_permission_overrides (user_id, bid, permission_code, allowed) VALUES (%s, %s, %s, 1)""", (user_id, str(bid), code), ) for code in sorted(base - desired): cursor.execute( """INSERT INTO user_permission_overrides (user_id, bid, permission_code, allowed) VALUES (%s, %s, %s, 0)""", (user_id, str(bid), code), ) conn.commit() return {'message': 'Permission overrides updated'}, 200 except Exception as e: conn.rollback() logger.error("Error syncing permission overrides: %s", e) return {'error': str(e)}, 500 finally: conn.close() def get_permissions_for_user(self, user_id, bid, role='user'): """Return role permissions plus user-level overrides for one business.""" role = self.normalize_role(role) conn = self.get_connection() try: cursor = conn.cursor() cursor.execute( "SELECT permission_code FROM role_permissions WHERE role = %s", (role,), ) permissions = {row['permission_code'] for row in (cursor.fetchall() or [])} cursor.execute( """SELECT permission_code, allowed FROM user_permission_overrides WHERE user_id = %s AND bid = %s""", (user_id, str(bid)), ) for row in cursor.fetchall() or []: code = row['permission_code'] if row.get('allowed'): permissions.add(code) else: permissions.discard(code) return sorted(permissions) except Exception as e: logger.warning("Could not load permissions for user=%s bid=%s: %s", user_id, bid, e) if role in self.ADMIN_ROLES: return ['calls.view_all', 'leads.view_all', 'analytics.view_all', 'users.view', 'users.create'] if role == 'agent': return ['calls.view_own', 'leads.view_own', 'analytics.view_own'] if role == 'manager': return ['calls.view_team', 'leads.view_team', 'analytics.view_team'] return ['calls.view_all', 'leads.view_all', 'analytics.view_all'] finally: conn.close() def get_agent_mappings(self, user_id, bid): conn = self.get_connection() try: cursor = conn.cursor() cursor.execute( """SELECT id, agent_name, agent_phone, agent_extension, is_primary, is_active FROM user_agent_mapping WHERE user_id = %s AND bid = %s AND is_active = 1 ORDER BY is_primary DESC, id ASC""", (user_id, str(bid)), ) return cursor.fetchall() or [] except Exception as e: logger.warning("Could not load agent mappings for user=%s bid=%s: %s", user_id, bid, e) return [] finally: conn.close() def get_group_mappings(self, user_id, bid): conn = self.get_connection() try: cursor = conn.cursor() cursor.execute( """SELECT id, groupname, is_active FROM user_group_mapping WHERE user_id = %s AND bid = %s AND is_active = 1 ORDER BY groupname ASC""", (user_id, str(bid)), ) return cursor.fetchall() or [] except Exception as e: logger.warning("Could not load group mappings for user=%s bid=%s: %s", user_id, bid, e) return [] finally: conn.close() def get_user_scope(self, user_id, bid, role='user'): role = self.normalize_role(role) if role in self.ADMIN_ROLES or role == 'user': return {'type': 'business'} if role == 'manager': groups = [row.get('groupname') for row in self.get_group_mappings(user_id, bid) if row.get('groupname')] agents = self.get_agent_mappings(user_id, bid) return { 'type': 'team', 'groupnames': groups, 'agent_names': [row.get('agent_name') for row in agents if row.get('agent_name')], 'agent_phones': [row.get('agent_phone') for row in agents if row.get('agent_phone')], 'agent_extensions': [row.get('agent_extension') for row in agents if row.get('agent_extension')], } if role == 'agent': agents = self.get_agent_mappings(user_id, bid) return { 'type': 'own_agent', 'agent_names': [row.get('agent_name') for row in agents if row.get('agent_name')], 'agent_phones': [row.get('agent_phone') for row in agents if row.get('agent_phone')], 'agent_extensions': [row.get('agent_extension') for row in agents if row.get('agent_extension')], } return {'type': 'business'} def replace_agent_mappings(self, user_id, bid, mappings): """Replace active agent mappings for a user/business.""" conn = self.get_connection() try: cursor = conn.cursor() cursor.execute( "UPDATE user_agent_mapping SET is_active = 0 WHERE user_id = %s AND bid = %s", (user_id, str(bid)), ) for idx, item in enumerate(mappings or []): if not isinstance(item, dict): continue agent_name = str(item.get('agent_name') or item.get('agentname') or '').strip() or None agent_phone = str(item.get('agent_phone') or item.get('phone') or '').strip() or None agent_extension = str(item.get('agent_extension') or item.get('extension') or '').strip() or None if not (agent_name or agent_phone or agent_extension): continue cursor.execute( """INSERT INTO user_agent_mapping (user_id, bid, agent_name, agent_phone, agent_extension, is_primary, is_active) VALUES (%s, %s, %s, %s, %s, %s, 1) ON DUPLICATE KEY UPDATE is_primary = VALUES(is_primary), is_active = 1""", (user_id, str(bid), agent_name, agent_phone, agent_extension, 1 if idx == 0 else 0), ) conn.commit() return {'message': 'Agent mappings updated'}, 200 except Exception as e: conn.rollback() logger.error("Error replacing agent mappings: %s", e) return {'error': str(e)}, 500 finally: conn.close() def replace_group_mappings(self, user_id, bid, groupnames): """Replace active group mappings for a user/business.""" conn = self.get_connection() try: cursor = conn.cursor() cursor.execute( "UPDATE user_group_mapping SET is_active = 0 WHERE user_id = %s AND bid = %s", (user_id, str(bid)), ) for groupname in groupnames or []: groupname = str(groupname or '').strip() if not groupname: continue cursor.execute( """INSERT INTO user_group_mapping (user_id, bid, groupname, is_active) VALUES (%s, %s, %s, 1) ON DUPLICATE KEY UPDATE is_active = 1""", (user_id, str(bid), groupname), ) conn.commit() return {'message': 'Group mappings updated'}, 200 except Exception as e: conn.rollback() logger.error("Error replacing group mappings: %s", e) return {'error': str(e)}, 500 finally: conn.close() def get_users_by_business(self, bid): """Get users assigned to a business with role/scope metadata.""" conn = self.get_connection() try: cursor = conn.cursor() cursor.execute( """SELECT bu.id, bu.username, bu.email, bu.full_name, bu.role AS global_role, bu.is_master, bu.is_active, bu.created_at, bu.last_login, uba.role FROM user_business_access uba JOIN business_users bu ON bu.id = uba.user_id WHERE uba.bid = %s ORDER BY CASE WHEN uba.role IN ('admin', 'business_admin') THEN 0 ELSE 1 END, bu.id ASC""", (str(bid),), ) users = cursor.fetchall() or [] for user in users: role = self.normalize_role(user.get('role')) user['role'] = role user['permissions'] = self.get_permissions_for_user(user.get('id'), bid, role) user['scope'] = self.get_user_scope(user.get('id'), bid, role) user['agent_mappings'] = self.get_agent_mappings(user.get('id'), bid) user['group_mappings'] = self.get_group_mappings(user.get('id'), bid) return users, 200 except Exception as e: logger.error("Error getting users by business: %s", e) return {'error': str(e)}, 500 finally: conn.close() def create_business(self, bid, name, description=None): """Create a new business (master admin only)""" bid = str(bid) conn = self.get_connection() created_tables = [] try: cursor = conn.cursor() # Check if business already exists cursor.execute("SELECT bid FROM businesses WHERE bid = %s", (bid,)) if cursor.fetchone(): return {'error': 'Business ID already exists'}, 409 schema = self.db_config['database'] suffixes = ['raw_calls', 'sarvamresponse', 'callanalytics'] templates = {} for suffix in suffixes: template = self._find_template_table(cursor, schema, suffix) if not template: return {'error': f'No template table found for {suffix}'}, 500 templates[suffix] = template for suffix in suffixes: table_name = f"{bid}_{suffix}" if self._table_exists(cursor, schema, table_name): return {'error': f'Table {table_name} already exists'}, 409 for suffix in suffixes: table_name = f"{bid}_{suffix}" template = templates[suffix] cursor.execute( f"CREATE TABLE {self._quote_identifier(table_name)} " f"LIKE {self._quote_identifier(template)}" ) created_tables.append(table_name) cursor.execute( """INSERT INTO businesses (bid, name, description, is_active) VALUES (%s, %s, %s, TRUE)""", (bid, name, description) ) conn.commit() return { 'bid': bid, 'name': name, 'description': description }, 201 except pymysql.IntegrityError: conn.rollback() return {'error': 'Business ID already exists'}, 409 except Exception as e: conn.rollback() logger.error(f"Error creating business: {str(e)}") if created_tables: try: cleanup_cursor = conn.cursor() for table_name in created_tables: cleanup_cursor.execute( f"DROP TABLE IF EXISTS {self._quote_identifier(table_name)}" ) conn.commit() except Exception as cleanup_error: logger.error(f"Error cleaning up tables: {str(cleanup_error)}") return {'error': str(e)}, 500 finally: conn.close() def get_all_businesses(self): """Get all businesses""" conn = self.get_connection() try: cursor = conn.cursor() cursor.execute( """SELECT bid, name, description, is_active, created_at FROM businesses ORDER BY name""" ) businesses = cursor.fetchall() return businesses, 200 except Exception as e: logger.error(f"Error getting businesses: {str(e)}") return {'error': str(e)}, 500 finally: conn.close() @staticmethod def _is_yyyy_mm_dd(value): if not value or not isinstance(value, str): return False s = value.strip() if len(s) != 10: return False parts = s.split("-") if len(parts) != 3: return False try: y, m, d = int(parts[0]), int(parts[1]), int(parts[2]) return 1 <= m <= 12 and 1 <= d <= 31 and 1970 <= y <= 2100 except ValueError: return False def get_activity_log( self, limit=100, offset=0, user_id=None, activity_type=None, action_code=None, bid=None, username_contains=None, description_contains=None, date_from=None, date_to=None, ): """Return structured audit rows + total count (for Audit Trail UI pagination).""" self.ensure_user_activity_log_audit_columns() conn = self.get_connection() try: cursor = conn.cursor() where_sql = "WHERE 1=1" params = [] if user_id is not None: try: uid = int(user_id) except (TypeError, ValueError): uid = None if uid is not None and uid > 0: where_sql += " AND al.user_id = %s" params.append(uid) if activity_type: where_sql += " AND al.activity_type = %s" params.append(str(activity_type).strip()) if action_code: where_sql += " AND al.action_code = %s" params.append(str(action_code).strip()) if bid is not None and str(bid).strip(): where_sql += " AND al.bid = %s" params.append(str(bid).strip()) if username_contains and str(username_contains).strip(): where_sql += " AND LOCATE(LOWER(%s), LOWER(al.username)) > 0" params.append(str(username_contains).strip()) if description_contains and str(description_contains).strip(): where_sql += " AND LOCATE(LOWER(%s), LOWER(al.description)) > 0" params.append(str(description_contains).strip()) if date_from and self._is_yyyy_mm_dd(date_from): where_sql += " AND DATE(al.created_at) >= %s" params.append(str(date_from).strip()) if date_to and self._is_yyyy_mm_dd(date_to): where_sql += " AND DATE(al.created_at) <= %s" params.append(str(date_to).strip()) count_sql = f"SELECT COUNT(*) AS cnt FROM user_activity_log al {where_sql}" cursor.execute(count_sql, list(params)) count_row = cursor.fetchone() or {} total = int(count_row.get("cnt") or 0) limit = max(1, min(int(limit or 100), 500)) offset = max(0, int(offset or 0)) query = f"""SELECT al.id, al.user_id, al.username, al.activity_type, al.description, al.ip_address, al.user_agent, al.created_at, al.bid, al.business_name, al.action_code, bu.email AS user_email, bu.full_name AS user_full_name, b.name AS business_join_name, user_businesses.business_names AS assigned_business_names FROM user_activity_log al LEFT JOIN business_users bu ON al.user_id = bu.id LEFT JOIN businesses b ON al.bid IS NOT NULL AND al.bid COLLATE utf8mb4_unicode_ci = b.bid LEFT JOIN ( SELECT uba.user_id, GROUP_CONCAT(b2.name ORDER BY b2.name SEPARATOR ', ') AS business_names FROM user_business_access uba LEFT JOIN businesses b2 ON uba.bid = b2.bid GROUP BY uba.user_id ) user_businesses ON al.user_id = user_businesses.user_id {where_sql} ORDER BY al.created_at DESC LIMIT %s OFFSET %s""" list_params = list(params) + [limit, offset] cursor.execute(query, list_params) logs = cursor.fetchall() or [] for row in logs: code = row.get("action_code") or self._infer_action_code(row.get("activity_type")) row["action_code_resolved"] = code name = row.get("business_name") or row.get("business_join_name") or row.get("assigned_business_names") if not name and row.get("bid"): name = str(row["bid"]) row["business_display_name"] = name or "—" return {"logs": logs, "total": total}, 200 except Exception as e: logger.error(f"Error getting activity log: {str(e)}") return {'error': str(e)}, 500 finally: conn.close() # ========================================================================= # EMBED / IFRAME INTEGRATION # ========================================================================= def ensure_embed_tables(self): """Create embed_api_keys table if it doesn't exist""" conn = self.get_connection() try: cursor = conn.cursor() cursor.execute(""" CREATE TABLE IF NOT EXISTS embed_api_keys ( id INT AUTO_INCREMENT PRIMARY KEY, api_key VARCHAR(64) UNIQUE NOT NULL, bid VARCHAR(20) NOT NULL, partner_name VARCHAR(100) NOT NULL, allowed_origins TEXT DEFAULT NULL, is_active TINYINT(1) DEFAULT 1, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, expires_at TIMESTAMP NULL DEFAULT NULL, last_used_at TIMESTAMP NULL DEFAULT NULL, INDEX idx_api_key (api_key), INDEX idx_bid (bid) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci """) conn.commit() logger.info("embed_api_keys table ensured") except Exception as e: logger.error(f"Error ensuring embed tables: {str(e)}") finally: conn.close() @staticmethod def generate_api_key(): """Generate a unique API key for embed access""" return "sk_embed_" + secrets.token_hex(24) def create_embed_api_key(self, bid, partner_name, allowed_origins=None, expires_at=None): """Create a new embed API key for a business""" conn = self.get_connection() try: cursor = conn.cursor() # Check if business exists cursor.execute("SELECT bid FROM businesses WHERE bid = %s", (bid,)) if not cursor.fetchone(): return {'error': 'Business not found'}, 404 api_key = self.generate_api_key() cursor.execute( """INSERT INTO embed_api_keys (api_key, bid, partner_name, allowed_origins, expires_at) VALUES (%s, %s, %s, %s, %s)""", (api_key, bid, partner_name, allowed_origins, expires_at) ) conn.commit() return { 'id': cursor.lastrowid, 'api_key': api_key, 'bid': bid, 'partner_name': partner_name, 'allowed_origins': allowed_origins, 'expires_at': str(expires_at) if expires_at else None }, 201 except Exception as e: conn.rollback() logger.error(f"Error creating embed API key: {str(e)}") return {'error': str(e)}, 500 finally: conn.close() def list_embed_api_keys(self, bid=None): """List all embed API keys, optionally filtered by bid""" conn = self.get_connection() try: cursor = conn.cursor() if bid: cursor.execute( """SELECT ek.id, ek.api_key, ek.bid, b.name as business_name, ek.partner_name, ek.allowed_origins, ek.is_active, ek.created_at, ek.expires_at, ek.last_used_at FROM embed_api_keys ek LEFT JOIN businesses b ON ek.bid = b.bid WHERE ek.bid = %s ORDER BY ek.created_at DESC""", (bid,) ) else: cursor.execute( """SELECT ek.id, ek.api_key, ek.bid, b.name as business_name, ek.partner_name, ek.allowed_origins, ek.is_active, ek.created_at, ek.expires_at, ek.last_used_at FROM embed_api_keys ek LEFT JOIN businesses b ON ek.bid = b.bid ORDER BY ek.created_at DESC""" ) keys = cursor.fetchall() # Mask API keys - only show last 8 chars for key in keys: full_key = key['api_key'] key['api_key_masked'] = '***' + full_key[-8:] del key['api_key'] return keys, 200 except Exception as e: logger.error(f"Error listing embed API keys: {str(e)}") return {'error': str(e)}, 500 finally: conn.close() def revoke_embed_api_key(self, key_id): """Revoke (deactivate) an embed API key""" conn = self.get_connection() try: cursor = conn.cursor() cursor.execute( "UPDATE embed_api_keys SET is_active = 0 WHERE id = %s", (key_id,) ) conn.commit() if cursor.rowcount == 0: return {'error': 'API key not found'}, 404 return {'message': 'API key revoked successfully'}, 200 except Exception as e: conn.rollback() logger.error(f"Error revoking embed API key: {str(e)}") return {'error': str(e)}, 500 finally: conn.close() def validate_api_key(self, api_key, bid): """Validate an embed API key for a specific business""" conn = self.get_connection() try: cursor = conn.cursor() cursor.execute( """SELECT id, api_key, bid, partner_name, allowed_origins, expires_at FROM embed_api_keys WHERE api_key = %s AND bid = %s AND is_active = 1""", (api_key, bid) ) key_record = cursor.fetchone() if not key_record: return None # Check expiry if key_record['expires_at'] and key_record['expires_at'] < datetime.utcnow(): return None # Update last_used_at cursor.execute( "UPDATE embed_api_keys SET last_used_at = NOW() WHERE id = %s", (key_record['id'],) ) conn.commit() return key_record except Exception as e: logger.error(f"Error validating API key: {str(e)}") return None finally: conn.close() def generate_embed_token(self, bid, partner_name, api_key_id): """Generate a short-lived JWT token for iframe embedding""" payload = { 'type': 'embed', 'bid': str(bid), 'partner_name': partner_name, 'api_key_id': api_key_id, 'exp': datetime.utcnow() + timedelta(hours=1), 'iat': datetime.utcnow() } token = jwt.encode(payload, self.jwt_secret, algorithm=self.jwt_algorithm) return token def validate_embed_token(self, token): """Validate an embed JWT token""" try: payload = jwt.decode(token, self.jwt_secret, algorithms=[self.jwt_algorithm]) if payload.get('type') != 'embed': return None return { 'bid': payload.get('bid'), 'partner_name': payload.get('partner_name'), 'api_key_id': payload.get('api_key_id') } except jwt.ExpiredSignatureError: logger.warning("Embed token expired") return None except jwt.InvalidTokenError: logger.warning("Invalid embed token") return None def require_auth(f): """Decorator to require authentication for routes""" @wraps(f) def decorated_function(*args, **kwargs): # Get token from Authorization header auth_header = request.headers.get('Authorization') if not auth_header or not auth_header.startswith('Bearer '): return jsonify({'error': 'Missing or invalid authorization header'}), 401 token = auth_header.replace('Bearer ', '') # Validate token from flask import current_app auth_handler = current_app.auth_handler user = auth_handler.validate_token(token) if not user: return jsonify({'error': 'Invalid or expired token'}), 401 # Add user to request context request.current_user = user return f(*args, **kwargs) return decorated_function def require_master(f): """Decorator to require master admin access""" @wraps(f) def decorated_function(*args, **kwargs): user = request.current_user if not user.get('is_master'): return jsonify({'error': 'Master admin access required'}), 403 return f(*args, **kwargs) return decorated_function def require_business_access(f): """Decorator to ensure user has access to the requested business""" @wraps(f) def decorated_function(*args, **kwargs): # This assumes the route has a 'bid' parameter bid = kwargs.get('bid') or request.view_args.get('bid') user = request.current_user # Master admins can access all businesses if user.get('is_master'): return f(*args, **kwargs) # Check if user has access to this business user_businesses = [b['bid'] for b in user.get('businesses', [])] if bid not in user_businesses: return jsonify({'error': 'Access denied to this business'}), 403 return f(*args, **kwargs) return decorated_function