# HBLink User Managment Server from flask import Flask, render_template_string, request, make_response, jsonify, render_template, Markup, flash, redirect, url_for, current_app from flask_sqlalchemy import SQLAlchemy from flask_user import login_required, UserManager, UserMixin, user_registered, roles_required from werkzeug.security import check_password_hash from flask_login import current_user, login_user, logout_user from wtforms import StringField, SubmitField import requests import base64, hashlib from dmr_utils3.utils import int_id, bytes_4 from config import * import ast import json import datetime, time from flask_babelex import Babel import libscrc import random from flask_mail import Message, Mail try: from gen_script_template import gen_script except: pass script_links = {} mmdvm_logins = [] ##def gen_passphrase(dmr_id): ## _new_peer_id = bytes_4(int(str(dmr_id)[:7])) ## b_list = create_app().get_burnlist() ## print(_new_peer_id) #### try: ## #if get_burnlist()[_new_peer_id] != 0: ## for ui in b_list: ## if b_list != 0: ## calc_passphrase = base64.b64encode(bytes.fromhex(str(hex(libscrc.ccitt((_new_peer_id) + get_burnlist()[_new_peer_id].to_bytes(2, 'big') + burn_int.to_bytes(2, 'big') + append_int.to_bytes(2, 'big') + bytes.fromhex(str(hex(libscrc.posix((_new_peer_id) + get_burnlist()[_new_peer_id].to_bytes(2, 'big') + burn_int.to_bytes(2, 'big') + append_int.to_bytes(2, 'big'))))[2:].zfill(8)))))[2:].zfill(4)) + (_new_peer_id) + get_burnlist()[_new_peer_id].to_bytes(2, 'big') + burn_int.to_bytes(2, 'big') + append_int.to_bytes(2, 'big') + bytes.fromhex(str(hex(libscrc.posix((_new_peer_id) + get_burnlist()[_new_peer_id].to_bytes(2, 'big') + burn_int.to_bytes(2, 'big') + append_int.to_bytes(2, 'big'))))[2:].zfill(8))) #### except: ## else: ## calc_passphrase = base64.b64encode(bytes.fromhex(str(hex(libscrc.ccitt((_new_peer_id) + append_int.to_bytes(2, 'big') + bytes.fromhex(str(hex(libscrc.posix((_new_peer_id) + append_int.to_bytes(2, 'big'))))[2:].zfill(8)))))[2:].zfill(4)) + (_new_peer_id) + append_int.to_bytes(2, 'big') + bytes.fromhex(str(hex(libscrc.posix((_new_peer_id) + append_int.to_bytes(2, 'big'))))[2:].zfill(8))) #### print(calc_passphrase) ## if use_short_passphrase == True: ## return str(calc_passphrase)[-9:-1] ## elif use_short_passphrase ==False: ## return str(calc_passphrase)[2:-1] # Query radioid.net for list of IDs def get_ids(callsign): try: url = "https://www.radioid.net" response = requests.get(url+"/api/dmr/user/?callsign=" + callsign) result = response.json() ## print(result) # id_list = [] id_list = {} f_name = result['results'][0]['fname'] l_name = result['results'][0]['surname'] try: city = str(result['results'][0]['city'] + ', ' + result['results'][0]['state'] + ', ' + result['results'][0]['country']) except: city = result['results'][0]['country'] for i in result['results']: id_list[i['id']] = 0 return str([id_list, f_name, l_name, city]) except: return str([{}, '', '', '']) # Return string in NATO phonetics def convert_nato(string): d_nato = { 'A': 'ALPHA', 'B': 'BRAVO', 'C': 'CHARLIE', 'D': 'DELTA', 'E': 'ECHO', 'F': 'FOXTROT', 'G': 'GOLF', 'H': 'HOTEL', 'I': 'INDIA', 'J': 'JULIETT','K': 'KILO', 'L': 'LIMA', 'M': 'MIKE', 'N': 'NOVEMBER','O': 'OSCAR', 'P': 'PAPA', 'Q': 'QUEBEC', 'R': 'ROMEO', 'S': 'SIERRA', 'T': 'TANGO', 'U': 'UNIFORM', 'V': 'VICTOR', 'W': 'WHISKEY', 'X': 'X-RAY', 'Y': 'YANKEE', 'Z': 'ZULU', '0': 'zero(0)', '1': 'one(1)', '2': 'two(2)', '3': 'three(3)', '4': 'four(4)', '5': 'five(5)', '6': 'six(6)', '7': 'seven(7)', '8': 'eight(8)', '9': 'nine(9)', 'a': 'alpha', 'b': 'bravo', 'c': 'charlie', 'd': 'delta', 'e': 'echo', 'f': 'foxtrot', 'g': 'golf', 'h': 'hotel', 'i': 'india', 'j': 'juliett','k': 'kilo', 'l': 'lima', 'm': 'mike', 'n': 'november','o': 'oscar', 'p': 'papa', 'q': 'quebec', 'r': 'romeo', 's': 'sierra', 't': 'tango', 'u': 'uniform', 'v': 'victor', 'w': 'whiskey', 'x': 'x-ray', 'y': 'yankee', 'z': 'Zulu'} ns = '' for c in string: try: ns = ns + d_nato[c] + ' ' except: ns = ns + c + ' ' return ns # Class-based application configuration class ConfigClass(object): from config import MAIL_SERVER, MAIL_PORT, MAIL_USE_SSL, MAIL_USE_TLS, MAIL_USERNAME, MAIL_PASSWORD, MAIL_DEFAULT_SENDER, USER_ENABLE_EMAIL, USER_ENABLE_USERNAME, USER_REQUIRE_RETYPE_PASSWORD, USER_ENABLE_CHANGE_USERNAME, USER_ENABLE_MULTIPLE_EMAILS, USER_ENABLE_CONFIRM_EMAIL, USER_ENABLE_REGISTER, USER_AUTO_LOGIN_AFTER_CONFIRM, USER_SHOW_USERNAME_DOES_NOT_EXIST """ Flask application config """ # Flask settings SECRET_KEY = secret_key # Flask-SQLAlchemy settings SQLALCHEMY_DATABASE_URI = db_location # File-based SQL database SQLALCHEMY_TRACK_MODIFICATIONS = False # Avoids SQLAlchemy warning # Flask-User settings USER_APP_NAME = title # Shown in and email templates and page footers USER_EMAIL_SENDER_EMAIL = MAIL_DEFAULT_SENDER USER_EDIT_USER_PROFILE_TEMPLATE = 'flask_user/edit_user_profile.html' # Setup Flask-User def create_app(): """ Flask application factory """ # Create Flask app load app.config mail = Mail() app = Flask(__name__) app.config.from_object(__name__+'.ConfigClass') # Initialize Flask-BabelEx babel = Babel(app) # Initialize Flask-SQLAlchemy db = SQLAlchemy(app) # Define the User data-model. # NB: Make sure to add flask_user UserMixin !!! class User(db.Model, UserMixin): __tablename__ = 'users' id = db.Column(db.Integer, primary_key=True) active = db.Column('is_active', db.Boolean(), nullable=False, server_default='1') # User authentication information. The collation='NOCASE' is required # to search case insensitively when USER_IFIND_MODE is 'nocase_collation'. username = db.Column(db.String(100, collation='NOCASE'), nullable=False, unique=True) password = db.Column(db.String(255), nullable=False, server_default='') email_confirmed_at = db.Column(db.DateTime()) email = db.Column(db.String(255, collation='NOCASE'), nullable=False, unique=True) # User information first_name = db.Column(db.String(100, collation='NOCASE'), nullable=False, server_default='') last_name = db.Column(db.String(100, collation='NOCASE'), nullable=False, server_default='') dmr_ids = db.Column(db.String(100, collation='NOCASE'), nullable=False, server_default='') city = db.Column(db.String(100, collation='NOCASE'), nullable=False, server_default='') notes = db.Column(db.String(100, collation='NOCASE'), nullable=False, server_default='') #Used for initial approval initial_admin_approved = db.Column('initial_admin_approved', db.Boolean(), nullable=False, server_default='1') # Define the relationship to Role via UserRoles roles = db.relationship('Role', secondary='user_roles') # Define the Role data-model class Role(db.Model): __tablename__ = 'roles' id = db.Column(db.Integer(), primary_key=True) name = db.Column(db.String(50), unique=True) # Define the UserRoles association table class UserRoles(db.Model): __tablename__ = 'user_roles' id = db.Column(db.Integer(), primary_key=True) user_id = db.Column(db.Integer(), db.ForeignKey('users.id', ondelete='CASCADE')) role_id = db.Column(db.Integer(), db.ForeignKey('roles.id', ondelete='CASCADE')) class BurnList(db.Model): __tablename__ = 'burn_list' dmr_id = db.Column(db.Integer(), unique=True, primary_key=True) version = db.Column(db.Integer(), primary_key=True) class AuthLog(db.Model): __tablename__ = 'auth_log' login_dmr_id = db.Column(db.Integer(), primary_key=True) login_time = db.Column(db.DateTime(), primary_key=True) peer_ip = db.Column(db.String(100, collation='NOCASE'), nullable=False, server_default='') server_name = db.Column(db.Integer(), primary_key=True) login_auth_method = db.Column(db.String(100, collation='NOCASE'), nullable=False, server_default='') portal_username = db.Column(db.String(100, collation='NOCASE'), nullable=False, server_default='') login_type = db.Column(db.String(100, collation='NOCASE'), nullable=False, server_default='') # Customize Flask-User class CustomUserManager(UserManager): # Override or extend the default login view method def login_view(self): """Prepare and process the login form.""" # Authenticate username/email and login authenticated users. safe_next_url = self._get_safe_next_url('next', self.USER_AFTER_LOGIN_ENDPOINT) safe_reg_next = self._get_safe_next_url('reg_next', self.USER_AFTER_REGISTER_ENDPOINT) # Immediately redirect already logged in users if self.call_or_get(current_user.is_authenticated) and self.USER_AUTO_LOGIN_AT_LOGIN: return redirect(safe_next_url) # Initialize form login_form = self.LoginFormClass(request.form) # for login.html register_form = self.RegisterFormClass() # for login_or_register.html if request.method != 'POST': login_form.next.data = register_form.next.data = safe_next_url login_form.reg_next.data = register_form.reg_next.data = safe_reg_next # Process valid POST if request.method == 'POST' and login_form.validate(): # Retrieve User user = None user_email = None if self.USER_ENABLE_USERNAME: # Find user record by username user = self.db_manager.find_user_by_username(login_form.username.data) # Find user record by email (with form.username) if not user and self.USER_ENABLE_EMAIL: user, user_email = self.db_manager.get_user_and_user_email_by_email(login_form.username.data) else: # Find user by email (with form.email) user, user_email = self.db_manager.get_user_and_user_email_by_email(login_form.email.data) #Add aditional message if not user.initial_admin_approved: flash('You account is waiting for approval from an administrator. See the Help page for more information. You will receive an email when your account is approved.', 'success') if user: # Log user in safe_next_url = self.make_safe_url(login_form.next.data) return self._do_login_user(user, safe_next_url, login_form.remember_me.data) # Render form self.prepare_domain_translations() template_filename = self.USER_LOGIN_AUTH0_TEMPLATE if self.USER_ENABLE_AUTH0 else self.USER_LOGIN_TEMPLATE return render_template(template_filename, form=login_form, login_form=login_form, register_form=register_form) #user_manager = UserManager(app, db, User) user_manager = CustomUserManager(app, db, User) # Create all database tables db.create_all() if not User.query.filter(User.username == 'admin').first(): user = User( username='admin', email='admin@no.reply', email_confirmed_at=datetime.datetime.utcnow(), password=user_manager.hash_password('admin'), initial_admin_approved = True, notes='Default admin account created during installation.' ) user.roles.append(Role(name='Admin')) user.roles.append(Role(name='User')) db.session.add(user) db.session.commit() # Query radioid.net for list of DMR IDs, then add to DB @user_registered.connect_via(app) def _after_user_registered_hook(sender, user, **extra): edit_user = User.query.filter(User.username == user.username).first() radioid_data = ast.literal_eval(get_ids(user.username)) edit_user.dmr_ids = str(radioid_data[0]) edit_user.first_name = str(radioid_data[1]) edit_user.last_name = str(radioid_data[2]) edit_user.city = str(radioid_data[3]) user_role = UserRoles( user_id=edit_user.id, role_id=2, ) db.session.add(user_role) if default_account_state == False: edit_user.active = default_account_state edit_user.initial_admin_approved = False db.session.commit() def gen_passphrase(dmr_id): _new_peer_id = bytes_4(int(str(dmr_id)[:7])) trimmed_id = int(str(dmr_id)[:7]) b_list = get_burnlist() print(b_list) burned = False for ui in b_list.items(): print(ui) #print(b_list) if ui[0] == trimmed_id: if ui[0] != 0: calc_passphrase = base64.b64encode(bytes.fromhex(str(hex(libscrc.ccitt((_new_peer_id) + b_list[trimmed_id].to_bytes(2, 'big') + burn_int.to_bytes(2, 'big') + append_int.to_bytes(2, 'big') + bytes.fromhex(str(hex(libscrc.posix((_new_peer_id) + b_list[trimmed_id].to_bytes(2, 'big') + burn_int.to_bytes(2, 'big') + append_int.to_bytes(2, 'big'))))[2:].zfill(8)))))[2:].zfill(4)) + (_new_peer_id) + b_list[trimmed_id].to_bytes(2, 'big') + burn_int.to_bytes(2, 'big') + append_int.to_bytes(2, 'big') + bytes.fromhex(str(hex(libscrc.posix((_new_peer_id) + b_list[trimmed_id].to_bytes(2, 'big') + burn_int.to_bytes(2, 'big') + append_int.to_bytes(2, 'big'))))[2:].zfill(8))) burned = True if burned == False: calc_passphrase = base64.b64encode(bytes.fromhex(str(hex(libscrc.ccitt((_new_peer_id) + append_int.to_bytes(2, 'big') + bytes.fromhex(str(hex(libscrc.posix((_new_peer_id) + append_int.to_bytes(2, 'big'))))[2:].zfill(8)))))[2:].zfill(4)) + (_new_peer_id) + append_int.to_bytes(2, 'big') + bytes.fromhex(str(hex(libscrc.posix((_new_peer_id) + append_int.to_bytes(2, 'big'))))[2:].zfill(8))) if use_short_passphrase == True: return str(calc_passphrase)[-9:-1] elif use_short_passphrase ==False: return str(calc_passphrase)[2:-1] def update_from_radioid(callsign): edit_user = User.query.filter(User.username == callsign).first() #edit_user.dmr_ids = str(ast.literal_eval(get_ids(callsign))[0]) radioid_dict = ast.literal_eval(get_ids(callsign))[0] db_id_dict = ast.literal_eval(edit_user.dmr_ids) new_id_dict = db_id_dict.copy() for i in radioid_dict.items(): if i[0] in db_id_dict: pass elif i[0] not in db_id_dict: new_id_dict[i[0]] = 0 edit_user.dmr_ids = str(new_id_dict) edit_user.first_name = str(ast.literal_eval(get_ids(callsign))[1]) edit_user.last_name = str(ast.literal_eval(get_ids(callsign))[2]) edit_user.city = str(ast.literal_eval(get_ids(callsign))[3]) db.session.commit() # The Home page is accessible to anyone @app.route('/') def home_page(): #content = Markup('Index') return render_template('index.html') #, markup_content = content) @app.route('/help') def help_page(): #content = Markup('Index') return render_template('help.html') @app.route('/generate_passphrase/pi-star', methods = ['GET']) @login_required def gen_pi_star(): try: u = current_user ## print(u.username) id_dict = ast.literal_eval(u.dmr_ids) #u = User.query.filter_by(username=user).first() ## print(user_id) ## print(request.args.get('mode')) ## if request.args.get('mode') == 'generated': content = '''
Pi-Star Instructions
1: Log into your Pi-Star device. rpi-rw
sudo su - Note: Link can be used only once. To run the script again, simply reload the page and paste a new command into the command line. ''' for i in id_dict.items(): #if i[1] == '': link_num = str(random.randint(1,99999999)).zfill(8) + str(time.time()) + str(random.randint(1,99999999)).zfill(8) script_links[i[0]] = link_num content = content + '''\nDMR ID: ''' + str(i[0]) + ''':
''' #else: # content = content + '''\n Error ''' content = content + '''\n |
''' except: content = Markup('No DMR IDs found or other error.') #return str(content) return render_template('flask_user_layout.html', markup_content = Markup(content)) @app.route('/generate_passphrase', methods = ['GET']) @login_required def gen(): #print(str(gen_passphrase(3153591))) #(int(i[0]))) try: #content = Markup('The HTML String') #user_id = request.args.get('user_id') u = current_user ## print(u.username) id_dict = ast.literal_eval(u.dmr_ids) #u = User.query.filter_by(username=user).first() ## print(user_id) ## print(request.args.get('mode')) ## if request.args.get('mode') == 'generated': #print(id_dict) content = '\n' for i in id_dict.items(): if isinstance(i[1], int) == True and i[1] != 0: link_num = str(random.randint(1,99999999)).zfill(8) + str(time.time()) + str(random.randint(1,99999999)).zfill(8) script_links[i[0]] = link_num #print(script_links) content = content + '''\n
Your passphrase for ''' + str(i[0]) + ''': Copy and paste: ''' + str(gen_passphrase(int(i[0]))) + ''' Phonetically spelled: ''' + convert_nato(str(gen_passphrase(int(i[0])))) + ''' |
''' elif i[1] == 0: link_num = str(random.randint(1,99999999)).zfill(8) + str(time.time()) + str(random.randint(1,99999999)).zfill(8) script_links[i[0]] = link_num #print(script_links) content = content + '''\n
Your passphrase for ''' + str(i[0]) + ''': Copy and paste: ''' + str(gen_passphrase(int(i[0]))) + ''' Phonetically spelled: ''' + convert_nato(str(gen_passphrase(int(i[0])))) + ''' |
''' elif i[1] == '': content = content + '''
Your passphrase for ''' + str(i[0]) + ''': Copy and paste: ''' + legacy_passphrase + ''' Phonetically spelled: ''' + convert_nato(legacy_passphrase) + ''' |
''' else: content = content + '''
Your passphrase for ''' + str(i[0]) + ''': Copy and paste: ''' + str(i[1]) + ''' Phonetically spelled: ''' + convert_nato(str(i[1])) + ''' |
''' #content = content + '\n\n' + str(script_links[i[0]]) except: content = Markup('No DMR IDs found or other error.') #return str(content) return render_template('view_passphrase.html', markup_content = Markup(content)) ## # The Members page is only accessible to authenticated users via the @login_required decorator ## @app.route('/members') ## @login_required # User must be authenticated ## def member_page(): ## content = 'Mem only' ## return render_template('flask_user_layout.html', markup_content = content) @app.route('/update_ids', methods=['POST', 'GET']) @login_required # User must be authenticated def update_info(): #print(request.args.get('callsign')) #print(current_user.username) if request.args.get('callsign') == current_user.username or request.args.get('callsign') and request.args.get('callsign') != current_user.username and current_user.has_roles('Admin'): content = '
Use this page to sync changes from RadioID.net with this system (such as a new DMR ID, name change, etc.).
Updating your information from RadioID.net will overwrite any custom authentication passphrases, your city, and name in the database. Are you sure you want to continue?
''' elif request.method == 'POST': # and request.form.get('callsign') and request.form.get('subject') and request.form.get('message'): u = User.query.filter_by(username=request.args.get('callsign')).first() msg = Message(recipients=[u.email], subject=request.form.get('subject'), body=request.form.get('message')) mail.send(msg) content = '
Sent email to: ' + u.email + '
' else: content = '''Find user in "List Users", then click on the email link.'
''' return render_template('flask_user_layout.html', markup_content = Markup(content)) @app.route('/list_users') @roles_required('Admin') @login_required # User must be authenticated def list_users(): u = User.query.all() # Broken for now, link taken out -u_list = '''
Callsign | Name | Enabled | DMR ID:Authentication | Notes |
''' + str(i.username) + ''' | ''' + str(i.first_name) + ' ' + str(i.last_name) + ''' | ''' + str(i.active) + ''' | ''' + str(i.dmr_ids) + ''' | ''' + str(i.notes) + ''' |
''' return render_template('flask_user_layout.html', markup_content = Markup(content)) @app.route('/approve_users', methods=['POST', 'GET']) @login_required @roles_required('Admin') # Use of @roles_required decorator def approve_list(): u = User.query.all() wait_list = '''
Callsign | Name | Enabled | DMR ID:Authentication |
''' + str(i.username) + ''' | ''' + str(i.first_name) + ' ' + str(i.last_name) + ''' | ''' + str(i.active) + ''' | ''' + str(i.dmr_ids) + ''' |
''' return render_template('flask_user_layout.html', markup_content = Markup(content)) # The Admin page requires an 'Admin' role. @app.route('/edit_user', methods=['POST', 'GET']) @login_required @roles_required('Admin') # Use of @roles_required decorator def admin_page(): #print(request.args.get('callsign')) #print(request.args.get('callsign')) ## if request.method == 'POST' and request.form.get('callsign'): ## #result = request.json ## callsign = request.form.get('callsign') ## u = User.query.filter_by(username=callsign).first() ## content = u.dmr_ids if request.method == 'POST' and request.args.get('callsign') == None: content = 'Not found' elif request.method == 'POST' and request.args.get('callsign') and request.form.get('user_status'): user = request.args.get('callsign') #print(user) edit_user = User.query.filter(User.username == user).first() content = '' if request.form.get('user_status') != edit_user.active: if request.form.get('user_status') == "True": edit_user.active = True content = content + '''
User ''' + str(user) + ''' has been enabled.
\n''' if request.form.get('user_status') == "False": edit_user.active = False content = content + '''User ''' + str(user) + ''' has been disabled.
\n''' ## print(request.form.get('username')) if user != request.form.get('username'): #### #print(edit_user.username) content = content + '''User ''' + str(user) + ''' changed to ''' + request.form.get('username') + '''.
\n''' edit_user.username = request.form.get('username') if request.form.get('email') != edit_user.email: edit_user.email = request.form.get('email') content = content + '''Changed email for user: ''' + str(user) + ''' to ''' + request.form.get('email') + '''
\n''' if request.form.get('notes') != edit_user.notes: edit_user.notes = request.form.get('notes') content = content + '''Changed notes for user: ''' + str(user) + '''.
\n''' if request.form.get('password') != '': edit_user.password = user_manager.hash_password(request.form.get('password')) content = content + '''Changed password for user: ''' + str(user) + '''
\n''' if request.form.get('dmr_ids') != edit_user.dmr_ids: edit_user.dmr_ids = request.form.get('dmr_ids') dmr_auth_dict = ast.literal_eval(request.form.get('dmr_ids')) for id_user in dmr_auth_dict: if isinstance(dmr_auth_dict[id_user], int) == True and dmr_auth_dict[id_user] != 0: #print('burn it') if id_user in get_burnlist(): ## print('burned') if get_burnlist()[id_user] != dmr_auth_dict[id_user]: ## print('update vers') update_burnlist(id_user, dmr_auth_dict[id_user]) else: pass ## print('no update') else: add_burnlist(id_user, dmr_auth_dict[id_user]) ## print('not in list, adding') elif isinstance(dmr_auth_dict[id_user], int) == False and id_user in get_burnlist(): delete_burnlist(id_user) ## print('remove from burn list - string') elif dmr_auth_dict[id_user] == 0: ## print('remove from burn list') if id_user in get_burnlist(): delete_burnlist(id_user) content = content + '''Changed authentication settings for user: ''' + str(user) + '''
\n''' db.session.commit() #edit_user = User.query.filter(User.username == request.args.get('callsign')).first() elif request.method == 'GET' and request.args.get('callsign') and request.args.get('delete_user') == 'true': delete_user = User.query.filter(User.username == request.args.get('callsign')).first() db.session.delete(delete_user) db.session.commit() content = '''Deleted user: ''' + str(delete_user.username) + '''
\n''' elif request.method == 'GET' and request.args.get('callsign') and request.args.get('make_user_admin') == 'true': u = User.query.filter_by(username=request.args.get('callsign')).first() u_role = UserRoles.query.filter_by(user_id=u.id).first() u_role.role_id = 1 db.session.commit() content = '''User now Admin: ''' + str(request.args.get('callsign')) + '''
\n''' elif request.method == 'GET' and request.args.get('callsign') and request.args.get('make_user_admin') == 'false': u = User.query.filter_by(username=request.args.get('callsign')).first() u_role = UserRoles.query.filter_by(user_id=u.id).first() u_role.role_id = 2 db.session.commit() content = '''Admin now a user: ''' + str(request.args.get('callsign') ) + '''
\n''' elif request.method == 'GET' and request.args.get('callsign') and request.args.get('admin_approve') == 'true': edit_user = User.query.filter(User.username == request.args.get('callsign')).first() edit_user.active = True edit_user.initial_admin_approved = True db.session.commit() msg = Message(recipients=[edit_user.email], subject='Account Approval - ' + title, body='''You are receiving this message because an administrator has approved your account. You may now login and view your MMDVM passphrase(s).''') mail.send(msg) content = '''User approved: ''' + str(request.args.get('callsign')) + '''
\n''' elif request.method == 'GET' and request.args.get('callsign') and request.args.get('email_verified') == 'true': edit_user = User.query.filter(User.username == request.args.get('callsign')).first() edit_user.email_confirmed_at = datetime.datetime.utcnow() db.session.commit() content = '''Email verified for: ''' + str(request.args.get('callsign')) + '''
\n''' elif request.method == 'POST' and request.form.get('callsign') and not request.form.get('user_status') or request.method == 'GET' and request.args.get('callsign'):# and request.form.get('user_status') : if request.args.get('callsign'): callsign = request.args.get('callsign') if request.form.get('callsign'): callsign = request.form.get('callsign') u = User.query.filter_by(username=callsign).first() confirm_link = '' if u.email_confirmed_at == None: confirm_link = '''Verify email - ''' + str(u.username) + '''
\n''' u_role = UserRoles.query.filter_by(user_id=u.id).first() if u_role.role_id == 2: # Link to promote to Admin role_link = '''Give Admin role: ''' + str(u.username) + '''
\n''' if u_role.role_id == 1: # Link to promote to User role_link = '''Revert to User role: ''' + str(u.username) + '''
\n''' id_dict = ast.literal_eval(u.dmr_ids) passphrase_list = '''DMR ID | Passphrase |
''' + str(i[0]) + ''' | ''' + str(gen_passphrase(int(i[0]))) + ''' |
''' + str(i[0]) + ''' | ''' + legacy_passphrase + ''' |
''' + str(i[0]) + ''' | ''' + str(i[1]) + ''' |
First Name | Last Name |
''' + u.first_name + ''' | ''' + u.last_name + ''' |
City | ''' + u.city + ''' |
''' + passphrase_list + '''
''' + confirm_link + ''' Email confirmed: ''' + str(u.email_confirmed_at) + ''' |
|
''' + role_link + ''' | |
Calculated | Legacy (config) | Custom |
0 - default, 1-999 - new calculation |
'' | 'passphrase' |
{DMR ID: Method, 2nd DMR ID: Method}
Example:
{1234567: '', 134568: 0, 1234569: 'passphr8s3'}
''' return render_template('flask_user_layout.html', markup_content = Markup(content)) @app.route('/get_script') def get_script(): dmr_id = int(request.args.get('dmr_id')) number = float(request.args.get('number')) #print(type(script_links[dmr_id])) u = User.query.filter(User.dmr_ids.contains(request.args.get('dmr_id'))).first() #print(u.dmr_ids) if authorized_peer(dmr_id)[1] == 0: passphrase = gen_passphrase(dmr_id) elif authorized_peer(dmr_id)[1] != 0 and isinstance(authorized_peer(dmr_id)[1], int) == True: passphrase = gen_passphrase(dmr_id) elif authorized_peer(dmr_id)[1] == '': passphrase = legacy_passphrase print(passphrase) elif authorized_peer(dmr_id)[1] != '' or authorized_peer(dmr_id)[1] != 0: passphrase = authorized_peer(dmr_id)[1] #try: if dmr_id in script_links and number == float(script_links[dmr_id]): script_links.pop(dmr_id) return str(gen_script(dmr_id, passphrase)) #except: #else: #content = 'Link used or other error.' #return content #return render_template('flask_user_layout.html', markup_content = content, logo = logo) def authorized_peer(peer_id): try: u = User.query.filter(User.dmr_ids.contains(str(peer_id))).first() login_passphrase = ast.literal_eval(u.dmr_ids) return [u.is_active, login_passphrase[peer_id], str(u.username)] except: return [False] @app.route('/auth_log', methods=['POST', 'GET']) @login_required # User must be authenticated @roles_required('Admin') def all_auth_list(): if request.args.get('flush_db') == 'true': content = '''
Flushed entire auth DB.
\n''' authlog_flush() elif request.args.get('flush_user_db') == 'true' and request.args.get('portal_username'): content = '''Flushed auth DB for: ''' + request.args.get('portal_username') + '''
\n''' authlog_flush_user(request.args.get('portal_username')) elif request.args.get('flush_db_mmdvm') == 'true' and request.args.get('mmdvm_server'): content = '''Flushed auth DB for: ''' + request.args.get('mmdvm_server') + '''
\n''' authlog_flush_mmdvm_server(request.args.get('mmdvm_server')) elif request.args.get('flush_db_ip') == 'true' and request.args.get('peer_ip'): content = '''Flushed auth DB for: ''' + request.args.get('peer_ip') + '''
\n''' authlog_flush_ip(request.args.get('peer_ip')) elif request.args.get('flush_dmr_id_db') == 'true' and request.args.get('dmr_id'): content = '''Flushed auth DB for: ''' + request.args.get('dmr_id') + '''
\n''' authlog_flush_dmr_id(request.args.get('dmr_id')) elif request.args.get('portal_username') and not request.args.get('flush_user_db') and not request.args.get('flush_dmr_id_db') or request.args.get('dmr_id') and not request.args.get('flush_user_db') and not request.args.get('flush_dmr_id_db'): if request.args.get('portal_username'): ## s_filter = portal_username=request.args.get('portal_username') a = AuthLog.query.filter_by(portal_username=request.args.get('portal_username')).order_by(AuthLog.login_time.desc()).all() g_arg = request.args.get('portal_username') f_link = '''Flush auth log for: ''' + request.args.get('portal_username') + '''
''' elif request.args.get('dmr_id'): ## s_filter = login_dmr_id=request.args.get('dmr_id') a = AuthLog.query.filter_by(login_dmr_id=request.args.get('dmr_id')).order_by(AuthLog.login_time.desc()).all() g_arg = request.args.get('dmr_id') f_link = '''Flush auth log for: ''' + request.args.get('dmr_id') + '''
''' ## print(s_filter) ## a = AuthLog.query.filter_by(s_filter).order_by(AuthLog.login_dmr_id.desc()).all() content = '''
Log for: ''' + g_arg + '''
''' + f_link + '''
DMR ID |
Portal Username |
Login IP |
Passphrase |
Server |
Time (UTC) |
Login Status |
''' + str(i.login_dmr_id) + ''' | ''' + i.portal_username + ''' | ''' + str(i.peer_ip) + ''' | ''' + i.login_auth_method + ''' | ''' + str(i.server_name) + ''' | ''' + str(i.login_time) + ''' | ''' + str(i.login_type) + ''' |
''' + str(i.login_dmr_id) + ''' | ''' + i.portal_username + ''' | ''' + str(i.peer_ip) + ''' | ''' + i.login_auth_method + ''' | ''' + str(i.server_name) + ''' | ''' + str(i.login_time) + ''' | ''' + str(i.login_type) + ''' |
''' + str(i.login_dmr_id) + ''' | ''' + i.portal_username + ''' | ''' + str(i.peer_ip) + ''' | ''' + i.login_auth_method + ''' | ''' + str(i.server_name) + ''' | ''' + str(i.login_time) + ''' | ''' + str(i.login_type) + ''' |
Flush authentication log for server: ''' + request.args.get('mmdvm_server') + '''
Log for MMDVM server: ''' + request.args.get('mmdvm_server') + '''
DMR ID |
Portal Username |
Login IP |
Passphrase |
Server |
Time (UTC) |
Login Status |
''' + str(i.login_dmr_id) + ''' | ''' + i.portal_username + ''' | ''' + str(i.peer_ip) + ''' | ''' + i.login_auth_method + ''' | ''' + i.server_name + ''' | ''' + str(i.login_time) + ''' | ''' + str(i.login_type) + ''' |
''' + str(i.login_dmr_id) + ''' | ''' + i.portal_username + ''' | ''' + str(i.peer_ip) + ''' | ''' + i.login_auth_method + ''' | ''' + i.server_name + ''' | ''' + str(i.login_time) + ''' | ''' + str(i.login_type) + ''' |
''' + str(i.login_dmr_id) + ''' | ''' + i.portal_username + ''' | ''' + str(i.peer_ip) + ''' | ''' + i.login_auth_method + ''' | ''' + i.server_name + ''' | ''' + str(i.login_time) + ''' | ''' + str(i.login_type) + ''' |
Flush authentication log for IP: ''' + request.args.get('peer_ip') + '''
Log for IP address: ''' + request.args.get('peer_ip') + '''
DMR ID |
Portal Username |
Login IP |
Passphrase |
Server |
Time (UTC) |
Login Status |
''' + str(i.login_dmr_id) + ''' | ''' + i.portal_username + ''' | ''' + i.peer_ip + ''' | ''' + i.login_auth_method + ''' | ''' + str(i.server_name) + ''' | ''' + str(i.login_time) + ''' | ''' + str(i.login_type) + ''' |
''' + str(i.login_dmr_id) + ''' | ''' + i.portal_username + ''' | ''' + i.peer_ip + ''' | ''' + i.login_auth_method + ''' | ''' + str(i.server_name) + ''' | ''' + str(i.login_time) + ''' | ''' + str(i.login_type) + ''' |
''' + str(i.login_dmr_id) + ''' | ''' + i.portal_username + ''' | ''' + i.peer_ip + ''' | ''' + i.login_auth_method + ''' | ''' + str(i.server_name) + ''' | ''' + str(i.login_time) + ''' | ''' + str(i.login_type) + ''' |
Flush entire authentication log
Un-registered authentication attempts
Authentication log by DMR ID
DMR ID |
Portal Username |
Login IP |
Passphrase |
Server |
Time (UTC) |
Last Login Status |
''' + str(i.login_dmr_id) + ''' | ''' + i.portal_username + ''' | ''' + str(i.peer_ip) + ''' | ''' + i.login_auth_method + ''' | ''' + str(i.server_name) + ''' | ''' + str(i.login_time) + ''' | ''' + str(i.login_type) + ''' |
''' + str(i.login_dmr_id) + ''' | ''' + i.portal_username + ''' | ''' + str(i.peer_ip) + ''' | ''' + i.login_auth_method + ''' | ''' + str(i.server_name) + ''' | ''' + str(i.login_time) + ''' | ''' + str(i.login_type) + ''' |
''' + str(i.login_dmr_id) + ''' | ''' + i.portal_username + ''' | ''' + str(i.peer_ip) + ''' | ''' + i.login_auth_method + ''' | ''' + str(i.server_name) + ''' | ''' + str(i.login_time) + ''' | ''' + str(i.login_type) + ''' |
''' elif request.method == 'POST' and request.form.get('username'): if not User.query.filter(User.username == request.form.get('username')).first(): radioid_data = ast.literal_eval(get_ids(request.form.get('username'))) user = User( username=request.form.get('username'), email=request.form.get('email'), email_confirmed_at=datetime.datetime.utcnow(), password=user_manager.hash_password(request.form.get('password')), dmr_ids = str(radioid_data[0]), initial_admin_approved = True, first_name = str(radioid_data[1]), last_name = str(radioid_data[2]), city = str(radioid_data[3]) ) db.session.add(user) u = User.query.filter_by(username=request.form.get('username')).first() user_role = UserRoles( user_id=u.id, role_id=2, ) db.session.add(user_role) db.session.commit() content = '''
Created user: ''' + str(request.form.get('username')) + '''
\n''' elif User.query.filter(User.username == request.form.get('username')).first(): content = 'Existing user: ' + str(request.form.get('username') + '. New user not created.') return render_template('flask_user_layout.html', markup_content = Markup(content)) @app.route('/auth', methods=['POST']) def auth(): hblink_req = request.json print((hblink_req)) if hblink_req['secret'] in shared_secrets: if 'login_id' in hblink_req and 'login_confirmed' not in hblink_req: if type(hblink_req['login_id']) == int: if authorized_peer(hblink_req['login_id'])[0]: if isinstance(authorized_peer(hblink_req['login_id'])[1], int) == True: authlog_add(hblink_req['login_id'], hblink_req['login_ip'], hblink_req['login_server'], authorized_peer(hblink_req['login_id'])[2], gen_passphrase(hblink_req['login_id']), 'Attempt') response = jsonify( allow=True, mode='normal', ) elif authorized_peer(hblink_req['login_id'])[1] == '': authlog_add(hblink_req['login_id'], hblink_req['login_ip'], hblink_req['login_server'], authorized_peer(hblink_req['login_id'])[2], 'Config Passphrase: ' + legacy_passphrase, 'Attempt') response = jsonify( allow=True, mode='legacy', ) elif authorized_peer(hblink_req['login_id'])[1] != '' or isinstance(authorized_peer(hblink_req['login_id'])[1], int) == False: authlog_add(hblink_req['login_id'], hblink_req['login_ip'], hblink_req['login_server'], authorized_peer(hblink_req['login_id'])[2], authorized_peer(hblink_req['login_id'])[1], 'Attempt') print(authorized_peer(hblink_req['login_id'])) response = jsonify( allow=True, mode='override', value=authorized_peer(hblink_req['login_id'])[1] ) elif authorized_peer(hblink_req['login_id'])[0] == False: print('log fail') authlog_add(hblink_req['login_id'], hblink_req['login_ip'], hblink_req['login_server'], 'Not Registered', '-', 'Failed') response = jsonify( allow=False) elif not type(hblink_req['login_id']) == int: user = hblink_req['login_id'] u = User.query.filter_by(username=user).first() if not u: msg = jsonify(auth=False, reason='User not found') response = make_response(msg, 401) if u: u_role = UserRoles.query.filter_by(user_id=u.id).first() password = user_manager.verify_password(hblink_req['password'], u.password) if u_role.role_id == 2: role = 'user' if u_role.role_id == 1: role = 'admin' if password: response = jsonify(auth=True, role=role) else: msg = jsonify(auth=False, reason='Incorrect password') response = make_response(msg, 401) elif 'login_id' in hblink_req and 'login_confirmed' in hblink_req: if hblink_req['old_auth'] == True: authlog_add(hblink_req['login_id'], hblink_req['login_ip'], hblink_req['login_server'], authorized_peer(hblink_req['login_id'])[2], 'CONFIG, NO UMS', 'Confirmed') else: authlog_add(hblink_req['login_id'], hblink_req['login_ip'], hblink_req['login_server'], authorized_peer(hblink_req['login_id'])[2], 'USER MANAGER', 'Confirmed') response = jsonify( logged=True ) elif hblink_req['burn_list']: # == 'burn_list': response = jsonify( burn_list=get_burnlist() ) else: message = jsonify(message='Authentication error') response = make_response(message, 401) return response return app if __name__ == '__main__': app = create_app() app.run(debug = True, port=ums_port, host=ums_host)