import traceback import argon2.exceptions def log_message(message): log_file = open("encsoft_log.txt", "a") log_file.write(message + "\n") log_file.close() try: import os import sys from flask import Flask, jsonify from flask_sqlalchemy import SQLAlchemy from flask import request, jsonify from datetime import datetime, timedelta from dateutil.relativedelta import relativedelta from argon2 import PasswordHasher, exceptions import pymysql allowed_software_versions = ["v1.0","v1.1","v1.2"] try: connection = pymysql.connect(host='localhost', user='softaimm_python_sc_access', password='yQ3hTdj5T45fVSa', database='softaimm_EncSoftDB', port=3306) log_message("Connection successful") connection.close() except pymysql.err.OperationalError as e: log_message("Connection failed: {}".format(e)) app = Flask(__name__) ph = PasswordHasher() # Check like this ph.verify(stored_hash, client_hash) # app.config["JWT_SECRET_KEY"] = "1d873c72dabb4251820273918d2a40698bebde9f53eb246c7b2f0b0b01a2d53b" # jwt = JWTManager(app) # Database connection details (replace with your actual values) db_user = 'softaimm_python_sc_access' # e.g., 'softaimm_user' db_pass = 'yQ3hTdj5T45fVSa' # The password from cPanel db_host = 'localhost' db_port = '3306' db_name = 'softaimm_EncSoftDB' # Use pymysql as the connector app.config['SQLALCHEMY_DATABASE_URI'] = f'mysql+pymysql://{db_user}:{db_pass}@{db_host}:{db_port}/{db_name}' app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False INTERNAL_SECRET = "ffdf9cf49f61679cbcd386904aa4fd8882b64a2cefadbac72d1b7a139438eb7c" db = SQLAlchemy(app) # Add this right after db = SQLAlchemy(app) to test connectivity with app.app_context(): try: db.engine.connect() log_message("Database connection successful!") except Exception as e: log_message("Connection failed: {}".format(e)) class ServerSettings(db.Model): __tablename__ = 'ServerSettings' id = db.Column(db.Integer, primary_key=True, autoincrement=True) login_expairy = db.Column(db.Integer, nullable=False) class Users(db.Model): __tablename__ = 'Users' id = db.Column(db.Integer, primary_key=True, autoincrement=True) username = db.Column(db.String(50), unique=True, nullable=False) email = db.Column(db.String(100), unique=True, nullable=False) password_hash = db.Column(db.String(255), nullable=False) created_at = db.Column(db.TIMESTAMP, server_default=db.func.current_timestamp()) updated_at = db.Column(db.TIMESTAMP, server_default=db.func.current_timestamp(), onupdate=db.func.current_timestamp()) last_login_at = db.Column(db.TIMESTAMP) account_status = db.Column(db.Enum('active', 'inactive', 'suspended'), default='active') allowed_devices = db.Column(db.Integer) allowed_groups = db.Column(db.Integer) software_version = db.Column(db.String(20)) notification = db.Column(db.Text) command = db.Column(db.String(50)) is_deleted = db.Column(db.Boolean, default=False, server_default='0', nullable=False) class Plans(db.Model): __tablename__ = 'Plans' plan_id = db.Column(db.Integer, primary_key=True, autoincrement=True) name = db.Column(db.String(50), nullable=False) description = db.Column(db.Text) price = db.Column(db.Numeric(10, 2), nullable=False) days = db.Column(db.Numeric(10, 2), nullable=False) duration_months = db.Column(db.Integer, nullable=False) per_month_price = db.Column(db.Numeric(10, 2), nullable=False) status = db.Column(db.Enum('active', 'inactive'), default='active') created_at = db.Column(db.TIMESTAMP, server_default=db.func.current_timestamp()) updated_at = db.Column(db.TIMESTAMP, server_default=db.func.current_timestamp(), onupdate=db.func.current_timestamp()) discount_percentage = db.Column(db.Numeric(5, 2)) def __repr__(self): return f'' class Subscriptions(db.Model): __tablename__ = 'Subscriptions' sub_id = db.Column(db.Integer, primary_key=True, autoincrement=True) product_key = db.Column(db.String(50), unique=True, nullable=False) registered_email = db.Column(db.String(255), unique=True, nullable=True) user_id = db.Column(db.Integer, db.ForeignKey('Users.id'), nullable=True) last_bought_plan = db.Column(db.Integer, nullable=True) start_date = db.Column(db.Date, nullable=True) end_date = db.Column(db.Date, nullable=True) status = db.Column(db.Enum('inactive','active', 'cancelled', 'expired'), default='active') created_at = db.Column(db.TIMESTAMP, server_default=db.func.current_timestamp()) updated_at = db.Column(db.TIMESTAMP, server_default=db.func.current_timestamp(), onupdate=db.func.current_timestamp()) def __repr__(self): return f'' class Devices(db.Model): __tablename__ = 'Devices' device_id = db.Column(db.Integer, primary_key=True, autoincrement=True) user_id = db.Column(db.Integer, db.ForeignKey('Users.id'), nullable=False) mcode_uuid = db.Column(db.String(50),nullable=True) mcode_serial = db.Column(db.String(50),nullable=True) hard_drive_code = db.Column(db.String(50),nullable=True) cpu_code = db.Column(db.String(50),nullable=True) changed_counter = db.Column(db.Integer) added_at = db.Column(db.TIMESTAMP, server_default=db.func.current_timestamp()) last_used_at = db.Column(db.TIMESTAMP) status = db.Column(db.Enum('active', 'inactive'), default='active') def __repr__(self): return f'' class LoginHistory(db.Model): __tablename__ = 'LoginHistory' id = db.Column(db.Integer, primary_key=True, autoincrement=True) user_id = db.Column(db.Integer, db.ForeignKey('Users.id'), nullable=False) login_at = db.Column(db.TIMESTAMP, server_default=db.func.current_timestamp()) success = db.Column(db.Text, nullable=False) # Consider using Boolean if it's just true/false ip_address = db.Column(db.String(45), nullable=True) user_agent = db.Column(db.String(255), nullable=True) # Relationship to Users table (optional) user = db.relationship('Users', backref='login_history') def __repr__(self): return f'' def get_user_account_details(user_id): """ Retrieve user details including their subscription information. Args: user_id (int): The ID of the user to look up Returns: tuple: (success: bool, result: dict/None) success indicates if operation was successful result contains data if successful, None otherwise """ try: # Get the user user = Users.query.filter_by(id=user_id).first() if not user: return (True, None) # User not found (this is a successful query) # Get active subscription (if any) active_subscription = Subscriptions.query.filter_by( user_id=user_id, status='active' ).first() # Prepare basic result result = { 'username': user.username, 'email': user.email, 'plan_name': None, 'expiry_date': None, 'groups_allowed':user.allowed_groups, 'devices_allowed':user.allowed_devices } if active_subscription: # Get plan details plan = Plans.query.get(active_subscription.last_bought_plan) if plan: result['plan_name'] = plan.name if active_subscription.end_date: result['expiry_date'] = active_subscription.end_date.isoformat() return (True, result) except Exception as e: log_message("Error in get_user_account_details: " + str(e)) tb_str = traceback.format_exception(etype=type(e), value=e, tb=e.__traceback__) log_message("".join(tb_str)) return (False, None) def get_login_expiry_hours(): server_settings = ServerSettings.query.filter_by(id=1).first() login_expiry_hours = server_settings.login_expairy if server_settings else None return login_expiry_hours def verify_username(username): """ Verify if a username is available (not already taken in the database). Args: username (str): The username to check Returns: bool: True if username is available, False if already taken """ # Query the Users table for the given username existing_user = Users.query.filter_by(username=username).first() # If no user found with this username, return True (available) # If user found, return False (already taken) return existing_user is None def verify_plan_id(plan_id): """ Verify if a plan_id exists in the database. Args: plan_id (int): The plan ID to check Returns: bool: True if plan exists, False if it doesn't """ plan_found = Plans.query.filter_by(plan_id=plan_id, status='active').first() return plan_found def get_user_id_by_username(username): """ Retrieve a user's ID by their username with error handling Args: username (str): The username to look up Returns: tuple: (success: bool, result: int/str) Where result is ID if success=True, or error message if success=False """ try: user = Users.query.filter_by(username=username).first() if user: return True, user.id return False, "User not found" except Exception as e: return False, f"Database error: {str(e)}" def get_plan_duration_months(plan_id): """ Retrieve the duration (in months) for a specific plan Args: plan_id (int): The ID of the plan to look up Returns: tuple: (success: bool, result: int/str) Where result is duration in months if success=True, or error message if success=False """ try: plan = Plans.query.get(plan_id) if plan: return True, plan.duration_months return False, "Plan not found" except Exception as e: return False, f"Database error: {str(e)}" def deactivate_user_account(username): """ Server Side Task 1. Sets the is_deleted field in users table to True (soft delete) 2. Cancels all active subscriptions for that user by setting status to 'cancelled' Args: username (str): The username of the user to deactivate Returns: tuple: (success: bool, message: str) """ try: # Find the user by username user = Users.query.filter_by(username=username).first() if not user: return False, "User not found" # Soft delete the user user.is_deleted = True # Cancel all active subscriptions for this user active_subscriptions = Subscriptions.query.filter_by( user_id=user.id, status='active' ).all() for subscription in active_subscriptions: subscription.status = 'cancelled' subscription.updated_at = db.func.current_timestamp() # Commit both changes as a single transaction db.session.commit() return True, (f"User '{username}' successfully deactivated. " f"Cancelled {len(active_subscriptions)} subscription(s).") except Exception as e: db.session.rollback() return False, f"Error deactivating user: {str(e)}" def is_product_key_available(product_key): """ Check if a product key exists and is not associated with any user Args: product_key (str): The product key to validate Returns: bool: True if key exists and has no user, False otherwise """ subscription = Subscriptions.query.filter_by(product_key=product_key).first() if subscription is not None and subscription.status == "inactive": return False, "Key is inactive" # we only proceed for active key if subscription is not None and subscription.user_id is None: return True, "Key is available" elif subscription is not None and subscription.user_id is not None: return False, "Key is already taken" else: return False, "Key Not Found" def add_user_record(username, email, password_hash, software_version=None, notification=None): """ Creates a new user in the database. Args: username (str): Unique username email (str): Unique email password_hash (str): Pre-hashed password software_version (str, optional): Version of the software notification (str, optional): Initial notification message Returns: tuple: (success: bool, result: Users or str) Where result is the user object if success=True, or an error message if success=False """ try: if Users.query.filter((Users.username == username) | (Users.email == email)).first(): return (False, "Username or email already exists") new_user = Users( username=username, email=email, allowed_devices=2, allowed_groups=2, password_hash=password_hash, account_status='active', is_deleted=False, software_version=software_version, notification=notification ) db.session.add(new_user) db.session.commit() return (True, new_user) except IntegrityError: db.session.rollback() return (False, "Username or email already exists") except Exception as e: db.session.rollback() return (False, f"An error occurred: {str(e)}") def get_plan_end_date(plan_id): """ Calculate the end date of a plan by adding its duration months to today's date. Args: plan_id (int): The ID of the plan to calculate end date for Returns: tuple: (success: bool, result: date/str) Where result is the end date if success=True, or error message if success=False """ # Get plan duration months plan_months_res, plan_months = get_plan_duration_months(plan_id) if not plan_months_res: return False, plan_months # Return the error message try: # Calculate end date today_date = datetime.today().date() plan_months_time = relativedelta(months=plan_months) end_date = today_date + plan_months_time return True, end_date except Exception as e: return False, f"Date calculation error: {str(e)}" def get_plan_id_by_product_key(product_key): """ Retrieve the plan_id associated with a product key Args: product_key (str): The product key to look up Returns: tuple: (success: bool, result: int/str) Where result is plan_id if success=True, or error message if success=False """ try: subscription = Subscriptions.query.filter_by(product_key=product_key).first() if subscription: return True, subscription.plan_id return False, "Product key not found" except Exception as e: return False, f"Database error: {str(e)}" def assign_subscription_to_user(product_key, user_id, end_date): """ Updates a subscription record with user_id and dates Args: product_key (str): The product key to update user_id (int): The user ID to assign end_date (date): The calculated end date of the subscription Returns: tuple: (success: bool, result: str/None) success=True if updated successfully, success=False if error occurred (result contains error message) """ try: # Get today's date once today = datetime.today().date() # Find and update the subscription in one atomic operation updated = Subscriptions.query.filter_by(product_key=product_key).update({ 'user_id': user_id, 'start_date': today, 'end_date': end_date, 'updated_at': datetime.now(), 'status': 'active' # Optional: ensure status is active }) if updated == 0: return False, "Product key not found" db.session.commit() return True, "User successfully subscribed" except Exception as e: db.session.rollback() return False, f"Database error: {str(e)}" def add_device_record(user_id, mcode_uuid=None, mcode_serial=None, hard_drive_code=None, cpu_code=None): """ Adds a new device record to the Devices table Args: user_id (int): The user ID this device belongs to mcode_uuid (str, optional): mcode_serial (str, optional): hard_drive_code (str, optional): Hard drive identifier cpu_code (str, optional): CPU identifier Returns: tuple: (success: bool, result: Devices/str) success=True with device object if created successfully, success=False with error message if failed """ try: current_time = datetime.now() new_device = Devices( user_id=user_id, mcode_uuid = mcode_uuid, mcode_serial = mcode_serial, hard_drive_code=hard_drive_code, cpu_code=cpu_code, changed_counter=0, added_at=current_time, last_used_at=current_time, status='active' ) db.session.add(new_device) db.session.commit() return True, new_device except Exception as e: db.session.rollback() return False, f"Error adding device: {str(e)}" def add_login_history(user_id, success, ip_address=None, user_agent=None): """ Records a login attempt in the LoginHistory table """ try: new_entry = LoginHistory( user_id=user_id, success=success, ip_address=ip_address, user_agent=user_agent ) db.session.add(new_entry) db.session.commit() return True, new_entry except Exception as e: db.session.rollback() return False, f"Error recording login history: {str(e)}" def get_user_record_for_username_password(username, password_entry): """ Fetch a user by username and password (plain text comparison). Args: username (str): Username to search for password (str): Plain-text password to match Returns: Users: The user SQLAlchemy model object if matched None: If no user found or password doesn't match """ user = Users.query.filter_by( username=username, is_deleted=False ).first() if user: try: log_message("hash :{}||||{}||||".format(user.password_hash,password_entry)) if ph.verify(user.password_hash, password_entry): return True, user # Returns None if no match except argon2.exceptions.VerifyMismatchError as e: log_message("Error in get_user_record_for_username_password: " + str(e)) tb_str = traceback.format_exception(etype=type(e), value=e, tb=e.__traceback__) log_message("".join(tb_str)) return False, "password mismatch error" except Exception as e: log_message("Error in get_user_record_for_username_password: " + str(e)) tb_str = traceback.format_exception(etype=type(e), value=e, tb=e.__traceback__) log_message("".join(tb_str)) return False, "error in get_user_record_for_username_password" else: return False, "username not found" def get_subscription_record_by_userid_and_productkey(user_id, product_key): """ Fetch a subscription by user_id and product_key. Args: user_id (int): The user's ID product_key (str): The product key to match Returns: Subscriptions: The subscription SQLAlchemy model object if matched None: If no subscription found """ subscription = Subscriptions.query.filter_by( user_id=user_id, product_key=product_key ).first() return subscription # Returns None if no match def get_subscription_record_by_userid(user_id): """ Fetch a subscription by user_id and product_key. Args: user_id (int): The user's ID product_key (str): The product key to match Returns: Subscriptions: The subscription SQLAlchemy model object if matched None: If no subscription found """ subscription = Subscriptions.query.filter_by( user_id=user_id, ).first() return subscription # Returns None if no match def get_active_devices_for_userid(user_id): """ Fetch all active devices for a user. Args: user_id (int): The user's ID Returns: List[Devices]: List of active Device objects (empty if none found) """ devices = Devices.query.filter_by( user_id=user_id, status='active' ).all() return devices # Returns empty list if no matches def get_active_plan_by_plan_id(plan_id): """ Fetch an active plan by its name. Args: plan_name (str): The name of the plan to search for Returns: Plans: The Plan object if found and active None: If no active plan matches the name """ plan = Plans.query.filter_by( plan_id=plan_id, status='active' ).first() return plan def updating_tables_for_successful_login(user_id, device_id, success_message): """ Adding a row in a login history devices - update row last_used_at where device_id matches login_history - add record with user_id, login_at, success users - update row last_login_at where user_id matches Returns: bool: True if all updates were successful """ current_time = datetime.utcnow() try: # 1. Update device last_used_at device = Devices.query.filter_by( device_id=device_id, user_id=user_id, status='active' ).first() if device: device.last_used_at = current_time # 2. Create login history record login_record = LoginHistory( user_id=user_id, login_at=current_time, success=success_message, ip_address=None, user_agent=None ) db.session.add(login_record) # 3. Update user last_login_at user = Users.query.get(user_id) if user: user.last_login_at = current_time db.session.commit() return True except Exception as e: db.session.rollback() # Log the error if needed print(f"Error updating login records: {str(e)}") return False def check_if_device_is_same(user_id, mcode_uuid, mcode_serial, hard_drive_code, cpu_code): # Query all active devices for the user devices = Devices.query.filter_by( user_id=user_id, status='active' ).all() # Prepare new codes dictionary new_codes = { 'mcode_uuid': mcode_uuid, 'mcode_serial': mcode_serial, 'hard_drive_code': hard_drive_code, 'cpu_code': cpu_code } for device in devices: # Create dictionary of device's non-None codes device_codes = { 'mcode_uuid': device.mcode_uuid, 'mcode_serial': device.mcode_serial, 'hard_drive_code': device.hard_drive_code, 'cpu_code': device.cpu_code } # Filter out None values from both device and new codes non_none_device_codes = {k: v for k, v in device_codes.items() if v is not None} non_none_new_codes = {k: v for k, v in new_codes.items() if v is not None} # Skip if no codes to compare if not non_none_device_codes: continue # Compare codes and count matches match_count = 0 for code_name, device_value in non_none_device_codes.items(): if code_name in non_none_new_codes and device_value == non_none_new_codes[code_name]: match_count += 1 # Calculate mismatch count total_compared = len(non_none_device_codes) mismatch_count = total_compared - match_count # Case 1: Full match (minimum 2 matches) if match_count >= 2 and mismatch_count == 0: # Update any None fields in device with new values updated = False for code_name in new_codes: if getattr(device, code_name) is None and new_codes[code_name] is not None: setattr(device, code_name, new_codes[code_name]) updated = True if updated: try: db.session.commit() except Exception as e: db.session.rollback() # Log error but still return the device log_message(f"Error updating device: {str(e)}") return True, device.device_id # Case 2: Zero match (no matches or >1 mismatch) elif match_count == 0 or mismatch_count > 1: # Skip this device continue # Case 3: One mismatch (exactly one mismatch) elif mismatch_count == 1: # Check changed counter if device.changed_counter >= 1: # Skip if already changed once or more continue # Update device with new values for None fields # whatever non none code is there in new codes needs to back propogate specially the changed one for code_name in new_codes: if new_codes[code_name] is not None: setattr(device, code_name, new_codes[code_name]) # Increment changed counter device.changed_counter += 1 try: db.session.commit() except Exception as e: db.session.rollback() # Log error but still return the device log_message(f"Error updating device: {str(e)}") return True, device.device_id # No matching device found return False, None # Test route @app.route('/') def index(): return "Database connection setup complete!" # Server-side: Flask route to handle the registration @app.route('/delete', methods=['POST']) def delete(): secret = request.headers.get('X-Internal-Secret') if not secret or secret != INTERNAL_SECRET: return jsonify({'success': False, 'message': 'Forbidden'}), 403 try: data = request.get_json() username = data.get('username') soft_delete_res, soft_delete_msg = deactivate_user_account(username) if soft_delete_res: return jsonify({"message": "User deleted successfully {}".format(soft_delete_msg)}), 200 else: return jsonify({"error": soft_delete_msg}), 400 except Exception as e: log_message("Error in Delete: " + str(e)) tb_str = traceback.format_exception(etype=type(e), value=e, tb=e.__traceback__) log_message("".join(tb_str)) # Server-side: Flask route to handle the registration @app.route('/product_key', methods=['POST']) def register_new_product_key_route(): secret = request.headers.get('X-Internal-Secret') if not secret or secret != INTERNAL_SECRET: return jsonify({'success': False, 'message': 'Forbidden'}), 403 try: data = request.get_json() product_keys = data.get('product_key') # Validate input if not product_keys or not isinstance(product_keys, list): return jsonify({ 'success': False, 'message': 'Missing or invalid product_keys list' }), 400 seen_keys = set() duplicates_in_request = set() keys_to_add = [] results = [] # Check for duplicates in request for key in product_keys: if key in seen_keys: duplicates_in_request.add(key) continue seen_keys.add(key) keys_to_add.append(key) # Check existing keys in database existing_keys = set() existing_subs = Subscriptions.query.filter( Subscriptions.product_key.in_(keys_to_add) ).all() if existing_subs: existing_keys = {sub.product_key for sub in existing_subs} # Process valid keys new_subscriptions = [] for key in keys_to_add: if key in existing_keys: results.append({ 'key': key, 'success': False, 'message': 'Key already exists in database' }) continue new_sub = Subscriptions( product_key=key, status='inactive' ) new_subscriptions.append(new_sub) results.append({ 'key': key, 'success': True, 'message': 'Added successfully', 'subscription_id': new_sub.sub_id }) # Bulk insert db.session.add_all(new_subscriptions) db.session.commit() # Add duplicate entries to results for key in duplicates_in_request: results.append({ 'key': key, 'success': False, 'message': 'Duplicate in request' }) return jsonify({ 'success': True, 'message': 'Batch processing complete', 'results': results, 'added_count': len(new_subscriptions), 'skipped_count': len(existing_keys) + len(duplicates_in_request) }), 201 except Exception as e: log_message("Error in Register: " + str(e)) tb_str = traceback.format_exception(etype=type(e), value=e, tb=e.__traceback__) log_message("".join(tb_str)) ## jump here # Category: ADMIN @app.route('/activate_product_key', methods=['POST']) def activate_product_key_route(): secret = request.headers.get('X-Internal-Secret') if not secret or secret != INTERNAL_SECRET: return jsonify({'success': False, 'message': 'Forbidden'}), 403 try: data = request.get_json() product_key = data.get('product_key') plan_id = data.get('plan_id') email = data.get('email') # Validate required fields if not product_key or not plan_id or not email: return jsonify({ 'success': False, 'message': 'Missing required fields (product_key, plan_id, email)' }), 400 # Find the subscription subscription = Subscriptions.query.filter_by(product_key=product_key).first() if not subscription: return jsonify({ 'success': False, 'message': 'Product key not found' }), 404 # Find the subscription subscription_duplicate_email = Subscriptions.query.filter_by(registered_email=email).first() if subscription_duplicate_email: return jsonify({ 'success': False, 'message': 'Email already registered with Product key :' + subscription_duplicate_email.product_key }), 401 # Case 1: Key has no user assigned (new activation) if subscription.user_id is None: # Find plan by plan_id plan = Plans.query.filter_by(plan_id=plan_id, status='active').first() if not plan: return jsonify({ 'success': False, 'message': 'Plan not found or inactive' }), 404 try: # Calculate dates today = datetime.utcnow().date() days_value = int(float(plan.days)) # Double conversion to handle Decimal end_date = today + timedelta(days=days_value) # Update subscription subscription.user_id = None subscription.last_bought_plan = plan_id subscription.start_date = today subscription.end_date = end_date subscription.registered_email = email subscription.status = 'active' db.session.commit() except Exception as days_error: log_message( f"ERROR: Failed to calculate end date. Days value: {plan.days}. Error: {str(days_error)}") log_message(f"TRACEBACK: {traceback.format_exc()}") return jsonify({ 'success': False, 'message': 'Invalid plan duration configuration' }), 400 return jsonify({ 'success': True, 'message': 'Product key activated successfully', 'subscription_id': subscription.sub_id, 'start_date': today.isoformat(), 'end_date': end_date.isoformat(), 'days_active': plan.days }), 200 # Case 2: Key already has a user assigned else: # Check if subscription is active if subscription.status == 'active': return jsonify({ 'success': False, 'message': 'Product key is already active' }), 400 # Check if subscription is expired today = datetime.utcnow().date() if subscription.end_date and subscription.end_date < today: return jsonify({ 'success': False, 'message': 'Product key has expired and cannot be activated' }), 400 # Reactivate inactive but not expired subscription subscription.status = 'active' db.session.commit() # Calculate days left days_left = (subscription.end_date - today).days return jsonify({ 'success': True, 'message': 'Product key reactivated successfully', 'subscription_id': subscription.sub_id, 'days_left': days_left, 'new_end_date': subscription.end_date.isoformat() }), 200 except SQLAlchemyError as e: log_message("Error in Register: " + str(e)) db.session.rollback() return jsonify({ 'success': False, 'message': f'Database error: {str(e)}' }), 500 except Exception as e: tb_str = traceback.format_exception(etype=type(e), value=e, tb=e.__traceback__) log_message("".join(tb_str)) return jsonify({ 'success': False, 'message': f'Unexpected error: {str(e)}' }), 500 # Category: INSOFT @app.route('/register_user_for_product_key', methods=['POST']) def register_new_user_for_product_key_route(): try: ''' Input: · Product key · Username · Password_hash · mcode_uuid mcode_serial · Hard drive code · CPU code ''' login_expiry_hours = get_login_expiry_hours() data = request.get_json() product_key = data.get('product_key') username = data.get("username") password_hash = data.get('password_hash') mcode_uuid = data.get("mcode_uuid") mcode_serial = data.get("mcode_serial") hard_drive_code = data.get("hard_drive_code") cpu_code = data.get("cpu_code") # email = data.get("email") software_version = data.get("software_version") notification = "Thanks for logging in" nonce = data.get("nonce") ''' Server Side Tasks · Validate the product key that its not associated with any user before · Add the record in the users table with the username, password_hash, account status (active), is_deleted (False) · Update Subscription table with that product key row with user_id, start_date (today), end_date according to plan months, updated_date (today) · Add the record in Devices table with user_id, mcode_uuid, mcode_serial, hard_drive_code, cpu_code, added_at (Today), last_used_at (Today), status (Active) · Add record in login history with user_id_login_at, success ''' product_key_available_res, product_key_available_msg = is_product_key_available(product_key) if not product_key_available_res: return jsonify({"error": product_key_available_msg,"nonce":"","login_expiry_hours":0, "login_date":""}), 400 # Find the subscription because of email subscription = Subscriptions.query.filter_by(product_key=product_key).first() if not subscription: return jsonify({ 'success': False, 'message': 'Product key not found' }), 404 add_user_record_res , new_user_record = add_user_record(username,subscription.registered_email,password_hash,software_version,notification) if not add_user_record_res and new_user_record == "Username or email already exists": return jsonify({"result": "ERROR : Add record error {}".format(new_user_record),"nonce":"", "login_expiry_hours":0,"login_date":""}), 401 # now that the user record is added successfully we will set the user_id in the subscription record # Update the subscription with the new user_id subscription.user_id = new_user_record.id subscription.updated_at = datetime.utcnow() # Update timestamp # Commit the changes to database db.session.commit() # now we add this device device_Record_res, device_Record = add_device_record(new_user_record.id,mcode_uuid,mcode_serial,hard_drive_code,cpu_code) if not device_Record: return jsonify({"result": "ERROR : Cannot add the device record","nonce":"", "login_expiry_hours":0,"login_date":""}), 400 # login history login_success = "New User Registered for Product Key Successfully" add_login_res, add_login_record = add_login_history(new_user_record.id,login_success) if not add_login_res: return jsonify({"result": "ERROR : Cannot add the login history record","nonce":"", "login_expiry_hours":0,"login_date":""}), 400 account_details_res , account_details = get_user_account_details(new_user_record.id) if account_details_res: account_details_username = account_details["username"] account_details_email = account_details["email"] account_details_plan_name = account_details["plan_name"] account_details_expiry_date = account_details["expiry_date"] account_details_groups_allowed = account_details["groups_allowed"] account_details_devices_allowed = account_details["devices_allowed"] return jsonify({"result": "SUCCESS : New user {} with ID {} Registered Successfully" "".format(new_user_record.id,new_user_record.username), "nonce":nonce,"login_expiry_hours":login_expiry_hours, "login_date": datetime.now().strftime("%Y-%m-%d"), "account_details_username":account_details_username, "account_details_email" : account_details_email, "account_details_plan_name" : account_details_plan_name, "account_details_expiry_date" : account_details_expiry_date, "account_details_groups_allowed" : account_details_groups_allowed, "account_details_devices_allowed" : account_details_devices_allowed}), 201 else: raise Exception("Error in Account Details") except Exception as e: log_message("Error in Register: " + str(e)) tb_str = traceback.format_exception(etype=type(e), value=e, tb=e.__traceback__) log_message("".join(tb_str)) # Category: INSOFT @app.route('/validate_product_key', methods=['POST']) def validate_product_key_route(): try: ''' Input: · Product key ''' data = request.get_json() product_key = data.get('product_key') nonce = data.get("nonce") ''' Server Side Tasks: · Takes the key and checks if this key is valid or not · Then it checks · Product Key already has a user o Product key user has reached the limit of devices § We tell the user to contact support o Product key has a room for one more device § We ask the user to enter username and password mentioning new device detected · Product key does not have a user o We register the new user for this product key ''' product_key_available_res, product_key_available_msg = is_product_key_available(product_key) # key is not even found if not product_key_available_res and product_key_available_msg == "Key Not Found": return jsonify({"result": "ERROR : {}".format(product_key_available_msg),"nonce":""}), 400 if not product_key_available_res and product_key_available_msg == "Key is inactive": return jsonify({"result": "ERROR : {}".format(product_key_available_msg),"nonce":""}), 401 # Key found but take with room for device if not product_key_available_res and product_key_available_msg == "Key is already taken": return jsonify({"result": "SUCCESS Key is found {}".format(product_key_available_msg), "nonce":nonce}), 200 # Key found and new user can be registered if product_key_available_res and product_key_available_msg == "Key is available": return jsonify({"result": "SUCCESS Key is found {}".format(product_key_available_msg), "nonce":nonce}), 201 except Exception as e: log_message("Error in Register: " + str(e)) tb_str = traceback.format_exception(etype=type(e), value=e, tb=e.__traceback__) log_message("".join(tb_str)) # Category: INSOFT @app.route('/old_user_new_device_login', methods=['POST']) def old_user_new_device_login_route(): try: ''' Input: · Product key · Username · Password_entry · Software_version mcode_uuid mcode_serial · Hard drive code · CPU code ''' login_expiry_hours = get_login_expiry_hours() data = request.get_json() product_key = data.get('product_key') username = data.get("username") password_entry = data.get('password_entry') mcode_uuid = data.get("mcode_uuid") mcode_serial = data.get("mcode_serial") hard_drive_code = data.get("hard_drive_code") cpu_code = data.get("cpu_code") software_version = data.get("software_version") nonce = data.get("nonce") ''' Server Side Tasks: · We first verify that this info is correct and all of the details (product key, username, password_hash) match for the user record · Then we check and identify if this login is from a new device or is it from the old device · Old device o You are good to go we send the necessary files to the user and login successfully · New device o Check if this user has the room for a new device then we login successfully after notifying the user that a new device is detected o Otherwise we tell the user to contact support and the login is Unsuccessful ''' # get the user_id from username and password_entry user_record_res, user_record = get_user_record_for_username_password(username,password_entry) if not user_record_res: return jsonify({"result": "ERROR : "+user_record, "nonce":"","login_date":"","login_expiry_hours":0}), 400 # now fetch the subscriber record for this user_id and product subscription_record = get_subscription_record_by_userid_and_productkey(user_record.id,product_key) if not subscription_record: return jsonify({"result": "ERROR : Subscriber record not found with user id and product key", "nonce":"","login_date":"","login_expiry_hours":0}), 400 if subscription_record.status != "active": return jsonify({"result": "ERROR : Subscriber record is not active", "nonce":"","login_date":"","login_expiry_hours":0}), 400 account_details_res , account_details = get_user_account_details(user_record.id) if account_details_res: account_details_username = account_details["username"] account_details_email = account_details["email"] account_details_plan_name = account_details["plan_name"] account_details_expiry_date = account_details["expiry_date"] account_details_groups_allowed = account_details["groups_allowed"] account_details_devices_allowed = account_details["devices_allowed"] else: raise Exception("Error in Account Details") # before we go ahead we check for software version and return 407 if old version if software_version not in allowed_software_versions: return jsonify({"result": "ERROR : Software Version Expired", "nonce": "", "login_expiry_hours": 0, "login_date": ""}), 407 # jump here # get the list of active devices for this user id devices = get_active_devices_for_userid(user_record.id) # meaning we couldn't find any device active if not devices: # This means all of the devices of this user are deactivated # so we can add this new device and Login Successful device_record_res, device_record = add_device_record(user_record.id,mcode_uuid,mcode_serial,hard_drive_code,cpu_code) if device_record_res: # now that the user has logged in successfully we need to update table updating_tables_res = updating_tables_for_successful_login(user_record.id,device_record.device_id,"SUCCESS : New Device Added (1) for user") if updating_tables_res: return jsonify({"result": "SUCCESS : New Device Added (1)", "nonce":nonce,"login_expiry_hours":login_expiry_hours, "login_date": datetime.now().strftime("%Y-%m-%d"), "account_details_username": account_details_username, "account_details_email": account_details_email, "account_details_plan_name": account_details_plan_name, "account_details_expiry_date": account_details_expiry_date, "account_details_groups_allowed":account_details_groups_allowed, "account_details_devices_allowed":account_details_devices_allowed }), 200 else: return jsonify({"result": "ERROR : Could not update the tables for successful login", "nonce":"","login_expiry_hours":0,"login_date":""}), 400 else: return jsonify({"result": "ERROR : Could not add the device record", "nonce":"","login_expiry_hours":0,"login_date":""}), 400 else: # now first we need to check the allowed devices for this user subscription plan_record = get_active_plan_by_plan_id(subscription_record.last_bought_plan) if not plan_record: return jsonify({"result": "ERROR : Plan record is not found for this subscription", "nonce":"","login_expiry_hours":0,"login_date":""}), 400 # but before we begin we need to check if this is the same device or different device same_device, same_device_id = check_if_device_is_same(user_record.id,mcode_uuid,mcode_serial,hard_drive_code,cpu_code) devices_allowed = user_record.allowed_devices if (devices_allowed > len(devices)) and not same_device: # this means we have a room for one more device # so we add the device for this user device_record_res, device_record = add_device_record(user_record.id, mcode_uuid,mcode_serial, hard_drive_code, cpu_code) if device_record_res: # now that the user has logged in successfully we need to update table updating_tables_res = updating_tables_for_successful_login(user_record.id, device_record.device_id, "SUCCESS : New Device Added ({}) for user".format(len(devices)+1)) if updating_tables_res: return jsonify({"result": "SUCCESS : New Device Added {}".format(len(devices)+1), "nonce":nonce,"login_expiry_hours":login_expiry_hours, "login_date": datetime.now().strftime("%Y-%m-%d"), "account_details_username": account_details_username, "account_details_email": account_details_email, "account_details_plan_name": account_details_plan_name, "account_details_expiry_date": account_details_expiry_date, "account_details_groups_allowed":account_details_groups_allowed, "account_details_devices_allowed":account_details_devices_allowed }), 201 else: return jsonify({"result": "ERROR : Could not update the tables for successful login", "nonce":"","login_expiry_hours":0,"login_date":""}), 400 else: return jsonify({"result": "ERROR : Could not add the device record", "nonce":"","login_expiry_hours":0,"login_date":""}), 400 elif same_device: updating_tables_res = updating_tables_for_successful_login(user_record.id, same_device_id, "SUCCESS : Old user logging in from same device") if updating_tables_res: return jsonify({"result": "SUCCESS : Old user logging in from same old active device", "nonce":nonce,"login_expiry_hours":login_expiry_hours, "login_date": datetime.now().strftime("%Y-%m-%d"), "account_details_username": account_details_username, "account_details_email": account_details_email, "account_details_plan_name": account_details_plan_name, "account_details_expiry_date": account_details_expiry_date, "account_details_groups_allowed":account_details_groups_allowed, "account_details_devices_allowed":account_details_devices_allowed }), 202 else: return jsonify({"result": "ERROR : Could not update the tables for successful login", "nonce":"","login_expiry_hours":0,"login_date":""}), 400 elif devices_allowed <= len(devices): return jsonify({"result": "ERROR : Device Limit Reached", "nonce":"","login_expiry_hours":0,"login_date":""}), 403 except Exception as e: log_message("Error in Register: " + str(e)) tb_str = traceback.format_exception(etype=type(e), value=e, tb=e.__traceback__) log_message("".join(tb_str)) # Category: INSOFT @app.route('/simple_login', methods=['POST']) def simple_login_route(): try: ''' Input: · Username · Password_entry · mcode_uuid mcode_serial · CPU_code · Hard_drive_code · Software_Version 200 : Same Device Login 201 : New Device Added (1) ANOMALY 202 : Simple Login New Devices Added 401 : Username or Password not found 402 : User subscriber record is not active 403 : Device Limit reached 400 : Exception ''' login_expiry_hours = get_login_expiry_hours() data = request.get_json() username = data.get("username") password_entry = data.get('password_entry') mcode_uuid = data.get("mcode_uuid") mcode_serial = data.get("mcode_serial") hard_drive_code = data.get("hard_drive_code") cpu_code = data.get("cpu_code") software_version = data.get("software_version") nonce = data.get("nonce") ''' Server Side Tasks: · We verify that we even have record for this Username and Password · Then we check if this user subscription is even active or not · Then we check if this device is even in the users already registered device . If no devices are found to be active . Then we add this device into table with ANOMALY message and Login Successfully · If its a same old device o then we let the user login successfully · If its a new device o Check if the user has room for a new device o If yes § We login user successfully with new device detected message o If No § We tell the user to contact support with Login Unsuccessful ''' # get the user_id from username and password_entry user_record_res, user_record = get_user_record_for_username_password(username,password_entry) if not user_record_res: return jsonify({"result": "ERROR : "+user_record, "nonce":"","login_date":"","login_expiry_hours":0}), 401 # now fetch the subscriber record for this user_id and product subscription_record = get_subscription_record_by_userid(user_record.id) if not subscription_record: return jsonify({"result": "ERROR : Subscriber record not found with user id and product key", "nonce":"","login_date":"","login_expiry_hours":0}), 400 if subscription_record.status != "active": return jsonify({"result": "ERROR : Subscriber record is not active", "nonce":"","login_date":"","login_expiry_hours":0}), 402 account_details_res , account_details = get_user_account_details(user_record.id) if account_details_res: account_details_username = account_details["username"] account_details_email = account_details["email"] account_details_plan_name = account_details["plan_name"] account_details_expiry_date = account_details["expiry_date"] account_details_groups_allowed = account_details["groups_allowed"] account_details_devices_allowed = account_details["devices_allowed"] else: raise Exception("Error in Account Details") # before we go ahead we check for software version and return 407 if old version if software_version not in allowed_software_versions: return jsonify({"result": "ERROR : Software Version Expired", "nonce": "", "login_expiry_hours": 0, "login_date": ""}), 407 # get the list of active devices for this user id devices = get_active_devices_for_userid(user_record.id) if not devices: # This means all of the devices of this user are deactivated but the user is trying to do # simple login we will allow the user to login successfully but add the message in the # login history as anomaly device_record_res, device_record = add_device_record(user_record.id,mcode_uuid, mcode_serial,hard_drive_code,cpu_code) if device_record_res: # now that the user has logged in successfully we need to update table updating_tables_res = updating_tables_for_successful_login(user_record.id,device_record.device_id,"SUCCESS : New Device Added (1) for user") if updating_tables_res: return jsonify({"result": "SUCCESS : ANOMALY (Simple Login) New Device Added (1)", "nonce":nonce,"login_expiry_hours":login_expiry_hours, "login_date": datetime.now().strftime("%Y-%m-%d"), "account_details_username": account_details_username, "account_details_email": account_details_email, "account_details_plan_name": account_details_plan_name, "account_details_expiry_date": account_details_expiry_date, "account_details_groups_allowed":account_details_groups_allowed, "account_details_devices_allowed":account_details_devices_allowed }), 201 else: return jsonify({"result": "ERROR : ANOMALY (Simple Login) Could not update the tables for successful login", "nonce":"","login_date":"","login_expiry_hours":0}), 400 else: return jsonify({"result": "ERROR : ANOMALY (Simple Login) Could not add the device record", "nonce":"","login_date":"","login_expiry_hours":0}), 400 else: # now first we need to check the allowed devices for this user subscription plan_record = get_active_plan_by_plan_id(subscription_record.last_bought_plan) if not plan_record: return jsonify({"result": "ERROR : Plan record is not found for this subscription", "nonce":"","login_date":"","login_expiry_hours":0}), 400 # but before we begin we need to check if this is the same device or different device same_device, same_device_id = check_if_device_is_same(user_record.id,mcode_uuid,mcode_serial,hard_drive_code,cpu_code) devices_allowed = user_record.allowed_devices if (devices_allowed > len(devices)) and not same_device: # this means we have a room for one more device # so we add the device for this user device_record_res, device_record = add_device_record(user_record.id, mcode_uuid,mcode_serial, hard_drive_code, cpu_code) if device_record_res: # now that the user has logged in successfully we need to update table updating_tables_res = updating_tables_for_successful_login(user_record.id, device_record.device_id, "SUCCESS : Simple Login New Device Added ({}) for user".format(len(devices)+1)) if updating_tables_res: return jsonify({"result": "SUCCESS : Simple Login New Device Added {}".format(len(devices)+1), "nonce":nonce,"login_expiry_hours":login_expiry_hours, "login_date": datetime.now().strftime("%Y-%m-%d"), "account_details_username": account_details_username, "account_details_email": account_details_email, "account_details_plan_name": account_details_plan_name, "account_details_expiry_date": account_details_expiry_date, "account_details_groups_allowed":account_details_groups_allowed, "account_details_devices_allowed":account_details_devices_allowed }), 202 else: return jsonify({"result": "ERROR : Simple Login Could not update the tables for successful login", "nonce":"","login_date":"","login_expiry_hours":0}), 400 else: return jsonify({"result": "ERROR : Simple Login Could not add the device record", "nonce":"","login_date":"","login_expiry_hours":0}), 400 elif same_device: updating_tables_res = updating_tables_for_successful_login(user_record.id, same_device_id, "SUCCESS : Simple Login user logging in from same device") if updating_tables_res: return jsonify({"result": "SUCCESS : Simple Login Successful with already active device", "nonce":nonce,"login_expiry_hours":login_expiry_hours, "login_date": datetime.now().strftime("%Y-%m-%d"), "account_details_username": account_details_username, "account_details_email": account_details_email, "account_details_plan_name": account_details_plan_name, "account_details_expiry_date": account_details_expiry_date, "account_details_groups_allowed":account_details_groups_allowed, "account_details_devices_allowed":account_details_devices_allowed }), 200 else: return jsonify({"result": "ERROR : Simple Login Could not update the tables for successful login", "nonce":"","login_date":"","login_expiry_hours":0}), 400 elif devices_allowed <= len(devices): return jsonify({"result": "ERROR : Simple Login Device Limit Reached", "nonce":"","login_date":"","login_expiry_hours":0}), 403 except Exception as e: log_message("Error in Register: " + str(e)) tb_str = traceback.format_exception(etype=type(e), value=e, tb=e.__traceback__) log_message("".join(tb_str)) # Category: ADMIN @app.route('/cancel_subscription', methods=['POST']) def cancel_user_subscription(): secret = request.headers.get('X-Internal-Secret') if not secret or secret != INTERNAL_SECRET: return jsonify({'success': False, 'message': 'Forbidden'}), 403 try: data = request.get_json() # Validate at least one identifier is provided if not any([data.get('username'), data.get('user_id'), data.get('product_key')]): return jsonify({ 'success': False, 'message': 'Must provide at least one identifier (username, user_id, or product_key)' }), 400 subscription = None user = None # Case 1: Product key provided (most direct) if 'product_key' in data and data['product_key']: subscription = Subscriptions.query.filter_by( product_key=data['product_key'] ).first() if not subscription: return jsonify({ 'success': False, 'message': 'No subscription found with provided product key' }), 404 # Case 2: Username provided elif 'username' in data and data['username']: user = Users.query.filter_by( username=data['username'], is_deleted=False ).first() if not user: return jsonify({ 'success': False, 'message': 'User not found with provided username' }), 404 subscription = Subscriptions.query.filter_by( user_id=user.id, status='active' ).first() if not subscription: return jsonify({ 'success': False, 'message': 'No active subscription found for this user' }), 404 # Case 3: User ID provided elif 'user_id' in data and data['user_id']: user = Users.query.filter_by( id=data['user_id'], is_deleted=False ).first() if not user: return jsonify({ 'success': False, 'message': 'User not found with provided user ID' }), 404 subscription = Subscriptions.query.filter_by( user_id=user.id, status='active' ).first() if not subscription: return jsonify({ 'success': False, 'message': 'No active subscription found for this user' }), 404 # Update subscription status subscription.status = 'cancelled' subscription.updated_at = datetime.utcnow() # Update user's allowed devices to 0 if they exist if subscription.user_id: user = Users.query.get(subscription.user_id) if user: user.allowed_devices = 0 user.updated_at = datetime.utcnow() db.session.commit() return jsonify({ 'success': True, 'message': 'Subscription cancelled successfully', 'subscription_id': subscription.sub_id, 'product_key': subscription.product_key, 'user_id': subscription.user_id, 'cancelled_at': datetime.utcnow().isoformat() }), 200 except SQLAlchemyError as e: db.session.rollback() return jsonify({ 'success': False, 'message': 'Database error while cancelling subscription', 'error': str(e) }), 500 except Exception as e: return jsonify({ 'success': False, 'message': 'Unexpected error while cancelling subscription', 'error': str(e) }), 500 # Category: ADMIN @app.route('/activate_subscription', methods=['POST']) def reactivate_subscription(): secret = request.headers.get('X-Internal-Secret') if not secret or secret != INTERNAL_SECRET: return jsonify({'success': False, 'message': 'Forbidden'}), 403 try: data = request.get_json() # Validate at least one identifier is provided if not any([data.get('username'), data.get('user_id'), data.get('product_key')]): return jsonify({ 'success': False, 'message': 'Must provide at least one identifier (username, user_id, or product_key)' }), 400 subscription = None user = None today = datetime.utcnow().date() # Case 1: Product key provided (most direct) if 'product_key' in data and data['product_key']: subscription = Subscriptions.query.filter_by( product_key=data['product_key'] ).first() if not subscription: return jsonify({ 'success': False, 'message': 'No subscription found with provided product key' }), 404 # Case 2: Username provided elif 'username' in data and data['username']: user = Users.query.filter_by( username=data['username'], is_deleted=False ).first() if not user: return jsonify({ 'success': False, 'message': 'User not found with provided username' }), 404 subscription = Subscriptions.query.filter_by( user_id=user.id ).first() if not subscription: return jsonify({ 'success': False, 'message': 'No subscription found for this user' }), 404 # Case 3: User ID provided elif 'user_id' in data and data['user_id']: user = Users.query.filter_by( id=data['user_id'], is_deleted=False ).first() if not user: return jsonify({ 'success': False, 'message': 'User not found with provided user ID' }), 404 subscription = Subscriptions.query.filter_by( user_id=user.id ).first() if not subscription: return jsonify({ 'success': False, 'message': 'No subscription found for this user' }), 404 # Check subscription status if subscription.status not in ['cancelled', 'expired']: return jsonify({ 'success': False, 'message': f'Subscription is {subscription.status} and cannot be reactivated', 'current_status': subscription.status }), 400 # Check expiry date if subscription.end_date and subscription.end_date < today: return jsonify({ 'success': False, 'message': 'Subscription has expired and cannot be reactivated', 'end_date': subscription.end_date.isoformat(), 'current_date': today.isoformat() }), 400 # Reactivate the subscription subscription.status = 'active' subscription.updated_at = datetime.utcnow() # Update user's allowed_devices if needed if subscription.user_id: user = user or Users.query.get(subscription.user_id) if user and user.allowed_devices == 0: user.allowed_devices = 1 # Or your default value user.updated_at = datetime.utcnow() db.session.commit() return jsonify({ 'success': True, 'message': 'Subscription reactivated successfully', 'subscription_id': subscription.sub_id, 'product_key': subscription.product_key, 'user_id': subscription.user_id, 'end_date': subscription.end_date.isoformat(), 'reactivated_at': datetime.utcnow().isoformat() }), 200 except SQLAlchemyError as e: db.session.rollback() return jsonify({ 'success': False, 'message': 'Database error while reactivating subscription', 'error': str(e) }), 500 except Exception as e: return jsonify({ 'success': False, 'message': 'Unexpected error while reactivating subscription', 'error': str(e) }), 500 # Category: ADMIN @app.route('/set_allowed_devices', methods=['POST']) def set_allowed_devices(): secret = request.headers.get('X-Internal-Secret') if not secret or secret != INTERNAL_SECRET: return jsonify({'success': False, 'message': 'Forbidden'}), 403 try: data = request.get_json() username = data.get('username') new_allowed_devices = data.get('allowed_devices') # Validate input if not username or new_allowed_devices is None: return jsonify({ 'success': False, 'message': 'Both username and allowed_devices are required' }), 400 if not isinstance(new_allowed_devices, int) or new_allowed_devices < 0: return jsonify({ 'success': False, 'message': 'allowed_devices must be a positive integer' }), 400 # Find the user user = Users.query.filter_by( username=username, is_deleted=False ).first() if not user: return jsonify({ 'success': False, 'message': 'User not found' }), 404 # Store old value for response old_allowed_devices = user.allowed_devices # Check if we need to deactivate devices (only when reducing count) if new_allowed_devices < old_allowed_devices: # Get active devices ordered by last_used_at (newest first) or added_at active_devices = Devices.query.filter_by( user_id=user.id, status='active' ).order_by( Devices.last_used_at.desc(), # Most recently used first Devices.added_at.desc() # Then by addition date ).all() # Calculate how many devices to deactivate devices_to_deactivate = len(active_devices) - new_allowed_devices if devices_to_deactivate > 0: # Deactivate the oldest/most inactive devices beyond the new limit for device in active_devices[new_allowed_devices:]: device.status = 'inactive' device.updated_at = datetime.utcnow() # Update the allowed_devices count user.allowed_devices = new_allowed_devices user.updated_at = datetime.utcnow() db.session.commit() return jsonify({ 'success': True, 'message': 'Allowed devices updated successfully', 'username': username, 'old_allowed_devices': old_allowed_devices, 'new_allowed_devices': new_allowed_devices, 'devices_deactivated': max(0, ( old_allowed_devices - new_allowed_devices)) if new_allowed_devices < old_allowed_devices else 0 }), 200 except SQLAlchemyError as e: db.session.rollback() return jsonify({ 'success': False, 'message': 'Database error while updating allowed devices', 'error': str(e) }), 500 except Exception as e: return jsonify({ 'success': False, 'message': 'Unexpected error while updating allowed devices', 'error': str(e) }), 500 # Category: ADMIN @app.route('/set_allowed_groups', methods=['POST']) def set_allowed_groups(): secret = request.headers.get('X-Internal-Secret') if not secret or secret != INTERNAL_SECRET: return jsonify({'success': False, 'message': 'Forbidden'}), 403 try: data = request.get_json() username = data.get('username') new_allowed_groups = data.get('allowed_groups') # Validate input if not username or new_allowed_groups is None: return jsonify({ 'success': False, 'message': 'Both username and allowed_groups are required' }), 400 if not isinstance(new_allowed_groups, int) or new_allowed_groups < 0: return jsonify({ 'success': False, 'message': 'allowed_groups must be a positive integer' }), 400 # Find the user user = Users.query.filter_by( username=username, is_deleted=False ).first() if not user: return jsonify({ 'success': False, 'message': 'User not found' }), 404 # Store old value for response old_allowed_groups = user.allowed_groups # Update the allowed_groups count user.allowed_groups = new_allowed_groups user.updated_at = datetime.utcnow() db.session.commit() return jsonify({ 'success': True, 'message': 'Allowed groups updated successfully', 'username': username, 'old_allowed_groups': old_allowed_groups, 'new_allowed_groups': new_allowed_groups }), 200 except SQLAlchemyError as e: db.session.rollback() return jsonify({ 'success': False, 'message': 'Database error while updating allowed groups', 'error': str(e) }), 500 except Exception as e: return jsonify({ 'success': False, 'message': 'Unexpected error while updating allowed groups', 'error': str(e) }), 500 # Category: ADMIN @app.route('/reset_devices', methods=['POST']) def reset_devices(): secret = request.headers.get('X-Internal-Secret') if not secret or secret != INTERNAL_SECRET: return jsonify({'success': False, 'message': 'Forbidden'}), 403 try: data = request.get_json() username = data.get('username') # Validate input if not username: return jsonify({ 'success': False, 'message': 'Username is required' }), 400 # Find the user user = Users.query.filter_by( username=username, is_deleted=False ).first() if not user: return jsonify({ 'success': False, 'message': 'User not found' }), 404 # Get all active devices for this user active_devices = Devices.query.filter_by( user_id=user.id, status='active' ).all() # If no active devices found if not active_devices: return jsonify({ 'success': True, 'message': 'No active devices found for this user', 'username': username, 'devices_reset': 0 }), 200 # Mark all devices as inactive for device in active_devices: device.status = 'inactive' device.updated_at = datetime.utcnow() # Update the user's allowed_devices to 0 user.allowed_devices = 0 user.updated_at = datetime.utcnow() db.session.commit() return jsonify({ 'success': True, 'message': f'Successfully reset {len(active_devices)} device(s)', 'username': username, 'devices_reset': len(active_devices), 'allowed_devices_updated': 0 }), 200 except SQLAlchemyError as e: db.session.rollback() return jsonify({ 'success': False, 'message': 'Database error while resetting devices', 'error': str(e) }), 500 except Exception as e: return jsonify({ 'success': False, 'message': 'Unexpected error while resetting devices', 'error': str(e) }), 500 # Category: ADMIN @app.route('/renew_subscription', methods=['POST']) def renew_subscription(): secret = request.headers.get('X-Internal-Secret') if not secret or secret != INTERNAL_SECRET: return jsonify({'success': False, 'message': 'Forbidden'}), 403 try: data = request.get_json() username = data.get('username') product_key = data.get('product_key') plan_id = data.get('plan_id') # Validate at least one identifier is provided if not username and not product_key: return jsonify({ 'success': False, 'message': 'Must provide at least one identifier (username or product_key)' }), 400 if not plan_id: return jsonify({ 'success': False, 'message': 'Plan ID is required' }), 400 # Find the plan plan = Plans.query.filter_by(plan_id=plan_id, status='active').first() if not plan: return jsonify({ 'success': False, 'message': 'Plan not found or inactive' }), 404 subscription = None user = None # Case 1: Both username and product_key provided if username and product_key: user = Users.query.filter_by( username=username, is_deleted=False ).first() if not user: return jsonify({ 'success': False, 'message': 'User not found' }), 404 subscription = Subscriptions.query.filter_by( product_key=product_key, user_id=user.id ).first() if not subscription: return jsonify({ 'success': False, 'message': 'No subscription found with this product key for the specified user' }), 404 # Case 2: Only username provided elif username: user = Users.query.filter_by( username=username, is_deleted=False ).first() if not user: return jsonify({ 'success': False, 'message': 'User not found' }), 404 subscription = Subscriptions.query.filter_by( user_id=user.id ).first() if not subscription: return jsonify({ 'success': False, 'message': 'No subscription found for this user' }), 404 # Case 3: Only product_key provided elif product_key: subscription = Subscriptions.query.filter_by( product_key=product_key ).first() if not subscription: return jsonify({ 'success': False, 'message': 'No subscription found with this product key' }), 404 # Calculate new end date today = datetime.utcnow().date() days_to_add = int(float(plan.days)) # Convert Numeric to int if subscription.end_date and subscription.end_date > today: # Case 1: Extend existing subscription new_end_date = subscription.end_date + timedelta(days=days_to_add) else: # Case 2: Create new subscription period new_end_date = today + timedelta(days=days_to_add) # Update subscription subscription.status = 'active' subscription.end_date = new_end_date subscription.last_bought_plan = plan.plan_id subscription.updated_at = datetime.utcnow() # Update user's allowed devices if needed if subscription.user_id: user = user or Users.query.get(subscription.user_id) if user and user.allowed_devices == 0: user.allowed_devices = 1 # Reset to default or your preferred value user.updated_at = datetime.utcnow() db.session.commit() return jsonify({ 'success': True, 'message': 'Subscription renewed successfully', 'subscription_id': subscription.sub_id, 'product_key': subscription.product_key, 'user_id': subscription.user_id, 'plan_id': plan.plan_id, 'new_end_date': new_end_date.isoformat(), 'days_added': days_to_add, 'renewed_at': datetime.utcnow().isoformat() }), 200 except SQLAlchemyError as e: db.session.rollback() return jsonify({ 'success': False, 'message': 'Database error while renewing subscription', 'error': str(e) }), 500 except Exception as e: return jsonify({ 'success': False, 'message': 'Unexpected error while renewing subscription', 'error': str(e) }), 500 @app.route('/fetch_user_details', methods=['POST']) def fetch_user_details(): data = request.get_json() username = data.get('username') product_key = data.get('product_key') # Validate input if not (username or product_key): return jsonify({ 'status': 'failed', 'message': 'Must provide either username or product key' }), 400 if username and product_key: return jsonify({ 'status': 'failed', 'message': 'Provide only one identifier (username OR product key)' }), 400 try: user = None subscription = None if username: # Find user by username user = Users.query.filter_by(username=username).first() if not user: return jsonify({ 'status': 'failed', 'message': f'User with username {username} not found' }), 404 if product_key: # Find subscription by product key subscription = Subscriptions.query.filter_by(product_key=product_key).first() if not subscription: return jsonify({ 'status': 'failed', 'message': f'Product key {product_key} not found' }), 404 # Get user from subscription if subscription.user_id: user = Users.query.get(subscription.user_id) if not user: return jsonify({ 'status': 'failed', 'message': f'User associated with product key {product_key} not found' }), 404 else: return jsonify({ 'status': 'failed', 'message': f'Product key {product_key} is not associated with any user' }), 400 # Get active devices active_devices = Devices.query.filter_by( user_id=user.id, status='active' ).all() # Get latest subscription if we didn't get it via product key if not subscription: subscription = Subscriptions.query.filter_by( user_id=user.id ).order_by(Subscriptions.start_date.desc()).first() # Build response result = { 'username': user.username, 'email': user.email, 'user_id': user.id, 'last_login': user.last_login_at.isoformat() if user.last_login_at else None, 'allowed_devices': user.allowed_devices, 'allowed_groups': user.allowed_groups, 'software_version': user.software_version, 'active_devices': [{ 'device_id': d.device_id, 'mcode_uuid': d.mcode_uuid, 'mcode_serial': d.mcode_serial, 'cpu_code': d.cpu_code, 'hard_drive_code': d.hard_drive_code } for d in active_devices], 'subscription_details': { 'product_key': subscription.product_key if subscription else None, 'status': subscription.status if subscription else None, 'expiry_date': subscription.end_date.isoformat() if subscription and subscription.end_date else None, 'last_bought_plan': subscription.last_bought_plan if subscription else None } } return jsonify({ 'status': 'success', 'data': result }) except SQLAlchemyError as e: return jsonify({ 'status': 'failed', 'message': f'Database error: {str(e)}' }), 500 except Exception as e: return jsonify({ 'status': 'failed', 'message': f'Unexpected error: {str(e)}' }), 500 # Category: ADMIN @app.route('/fetch_all_available_keys', methods=['POST']) def fetch_all_available_keys(): secret = request.headers.get('X-Internal-Secret') if not secret or secret != INTERNAL_SECRET: return jsonify({'success': False, 'message': 'Forbidden'}), 403 try: # Get ONLY the product_key field for available keys available_keys = Subscriptions.query.with_entities(Subscriptions.product_key).filter( Subscriptions.registered_email.is_(None), Subscriptions.user_id.is_(None), Subscriptions.last_bought_plan.is_(None), Subscriptions.start_date.is_(None), Subscriptions.end_date.is_(None) ).all() # Extract just the strings from the results keys_list = [key[0] for key in available_keys] # key[0] because it's a tuple return jsonify({ 'status': 'success', 'available_keys': keys_list # Just plain strings }) except Exception as e: return jsonify({ 'status': 'failed', 'message': str(e) }), 500 # Category: ADMIN @app.route('/issue_product_key', methods=['POST']) def issue_product_key(): # ── Verify internal secret ──────────────────────────────── secret = request.headers.get('X-Internal-Secret') if not secret or secret != INTERNAL_SECRET: return jsonify({'success': False, 'message': 'Forbidden'}), 403 try: data = request.get_json() plan_id = data.get('plan_id') email = data.get('email') if not plan_id or not email: return jsonify({ 'success': False, 'message': 'Missing required fields (plan_id, email)' }), 400 # ── Check email not already registered ──────────────── existing = Subscriptions.query.filter_by(registered_email=email).first() if existing: return jsonify({ 'success': False, 'message': 'Email already has an active subscription' }), 400 # ── Find one available unassigned key ───────────────── available_key = Subscriptions.query.filter( Subscriptions.registered_email.is_(None), Subscriptions.user_id.is_(None), Subscriptions.last_bought_plan.is_(None), Subscriptions.start_date.is_(None), Subscriptions.end_date.is_(None) ).first() if not available_key: return jsonify({ 'success': False, 'message': 'No available keys in pool' }), 503 # ── Find plan ───────────────────────────────────────── plan = Plans.query.filter_by(plan_id=plan_id, status='active').first() if not plan: return jsonify({ 'success': False, 'message': 'Plan not found or inactive' }), 404 # ── Activate the key ────────────────────────────────── today = datetime.utcnow().date() days_value = int(float(plan.days)) end_date = today + timedelta(days=days_value) available_key.last_bought_plan = plan_id available_key.start_date = today available_key.end_date = end_date available_key.registered_email = email available_key.status = 'active' db.session.commit() return jsonify({ 'success': True, 'message': 'Product key issued successfully', 'product_key': available_key.product_key, 'start_date': today.isoformat(), 'end_date': end_date.isoformat(), 'days_active': plan.days }), 200 except SQLAlchemyError as e: db.session.rollback() return jsonify({ 'success': False, 'message': f'Database error: {str(e)}' }), 500 except Exception as e: return jsonify({ 'success': False, 'message': f'Unexpected error: {str(e)}' }), 500 # Category: ADMIN @app.route('/verify_user', methods=['POST']) def verify_user_route(): secret = request.headers.get('X-Internal-Secret') if not secret or secret != INTERNAL_SECRET: return jsonify({'success': False, 'message': 'Forbidden'}), 403 try: ''' Input: · Username · Password_entry - Nonce 200 : Username and Password matched User Found 401 : Username or Password not found 400 : Exception ''' # login_expiry_hours = get_login_expiry_hours() data = request.get_json() username = data.get("username") password_entry = data.get('password_entry') nonce = data.get("nonce") ''' Server Side Tasks: · We verify that we even have record for this Username and Password · Then we check if this user subscription is even active or not · Then we check if this device is even in the users already registered device . If no devices are found to be active . Then we add this device into table with ANOMALY message and Login Successfully · If its a same old device o then we let the user login successfully · If its a new device o Check if the user has room for a new device o If yes § We login user successfully with new device detected message o If No § We tell the user to contact support with Login Unsuccessful ''' # get the user_id from username and password_entry user_record_res, user_record = get_user_record_for_username_password(username, password_entry) if not user_record_res: return jsonify({"result": "ERROR : " + user_record, "nonce": ""}), 401 return jsonify({"result": "SUCCESS : User Verify SUccessfully", "nonce": nonce}), 200 except Exception as e: log_message("Error in Register: " + str(e)) tb_str = traceback.format_exception(etype=type(e), value=e, tb=e.__traceback__) log_message("".join(tb_str)) return jsonify({"result": "ERROR : Server Error Occurred", "nonce": ""}), 400 if __name__ == '__main__': app.run() except Exception as e: log_message("An error occurred: " + str(e)) tb_str = traceback.format_exception(etype=type(e), value=e, tb=e.__traceback__) log_message("".join(tb_str))