CIRCL
/
AIL-framework
mirror of https://github.com/CIRCL/AIL-framework
2
1
Fork 0
Code Issues Releases Wiki Activity

chg: [UI] replace remote_addr by access_route

otp
terrtia 2024-09-06 11:16:52 +02:00
parent 23f92e0290
commit a51347bf3c
No known key found for this signature in database
GPG Key ID: 1E1B1F50D84613D0
3 changed files with 25 additions and 25 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
var/www/blueprints/api_rest.py
View File

@ -61,7 +61,7 @@ def token_required(user_role):
return create_json_response({'status': 'error', 'reason': 'Invalid Role'}, 401)
token = get_auth_from_header()
ip_source = request.remote_addr
ip_source = request.access_route[0]
data, status_code = ail_api.authenticate_user(token, ip_address=ip_source)
if status_code != 200:
return create_json_response(data, status_code)

20
var/www/blueprints/root.py
View File

@ -54,7 +54,7 @@ root = Blueprint('root', __name__, template_folder='templates')
# ============= ROUTES ==============
@root.route('/login', methods=['POST', 'GET']) # TODO LOG BRUTEFORCE ATTEMPT
def login():
current_ip = request.remote_addr
current_ip = request.access_route[0]
login_failed_ip = r_cache.get(f'failed_login_ip:{current_ip}')
# brute force by IP
@ -138,7 +138,7 @@ def login():
r_cache.expire(f'failed_login_user_id:{username}', 300)
#
access_logger.info(f'Login Failed', extra={'user_id': user.get_user_id(), 'ip_address': request.remote_addr})
access_logger.info(f'Login Failed', extra={'user_id': user.get_user_id(), 'ip_address': request.access_route[0]})
logging_error = 'Login/Password Incorrect'
return render_template("login.html", error=logging_error)
@ -166,7 +166,7 @@ def verify_2fa():
if otp_expire < int(time.time()): # TODO LOG
session.pop('user_id', None)
session.pop('otp_expire', None)
access_logger.info(f'First Login Expired', extra={'user_id': user_id, 'ip_address': request.remote_addr})
access_logger.info(f'First Login Expired', extra={'user_id': user_id, 'ip_address': request.access_route[0]})
error = "First Login Expired"
return redirect(url_for('root.login', error=error))
@ -188,7 +188,7 @@ def verify_2fa():
login_user(user)
user.update_last_login()
access_logger.info(f'2FA login', extra={'user_id': user.get_user_id(), 'ip_address': request.remote_addr})
access_logger.info(f'2FA login', extra={'user_id': user.get_user_id(), 'ip_address': request.access_route[0]})
if user.request_password_change():
return redirect(url_for('root.change_password'))
@ -199,7 +199,7 @@ def verify_2fa():
return redirect(url_for('dashboard.index'))
else:
htop_counter = user.get_htop_counter()
access_logger.info(f'Invalid OTP', extra={'user_id': user.get_user_id(), 'ip_address': request.remote_addr})
access_logger.info(f'Invalid OTP', extra={'user_id': user.get_user_id(), 'ip_address': request.access_route[0]})
error = "The OTP is incorrect or has expired"
return render_template("verify_otp.html", htop_counter=htop_counter, next_page=next_page, error=error)
@ -220,7 +220,7 @@ def setup_2fa():
if otp_expire < int(time.time()): # TODO LOG
session.pop('user_id', None)
session.pop('otp_expire', None)
access_logger.info(f'First Login Expired', extra={'user_id': user_id, 'ip_address': request.remote_addr})
access_logger.info(f'First Login Expired', extra={'user_id': user_id, 'ip_address': request.access_route[0]})
error = "First Login Expired"
return redirect(url_for('root.login', error=error))
@ -243,14 +243,14 @@ def setup_2fa():
login_user(user)
user.update_last_login()
access_logger.info(f'2FA login', extra={'user_id': user.get_user_id(), 'ip_address': request.remote_addr})
access_logger.info(f'2FA login', extra={'user_id': user.get_user_id(), 'ip_address': request.access_route[0]})
if user.request_password_change():
return redirect(url_for('root.change_password'))
else:
return redirect(url_for('dashboard.index'))
else:
access_logger.info(f'OTP Invalid', extra={'user_id': user.get_user_id(), 'ip_address': request.remote_addr})
access_logger.info(f'OTP Invalid', extra={'user_id': user.get_user_id(), 'ip_address': request.access_route[0]})
error = "The OTP is incorrect or has expired"
return redirect(url_for('root.setup_2fa', error=error))
else:
@ -278,7 +278,7 @@ def change_password():
res = api_change_user_self_password(user_id, password1)
if res[1] != 200:
return create_json_response(res[0], res[1])
access_logger.info(f'Password change', extra={'user_id': user_id, 'ip_address': request.remote_addr})
access_logger.info(f'Password change', extra={'user_id': user_id, 'ip_address': request.access_route[0]})
# update Note
# dashboard
return redirect(url_for('dashboard.index', update_note=True))
@ -295,7 +295,7 @@ def change_password():
@root.route('/logout')
@login_required
def logout():
access_logger.info(f'Logout', extra={'user_id': current_user.get_user_id(), 'ip_address': request.remote_addr})
access_logger.info(f'Logout', extra={'user_id': current_user.get_user_id(), 'ip_address': request.access_route[0]})
current_user.kill_session()
logout_user()
return redirect(url_for('root.login'))

28
var/www/blueprints/settings_b.py
View File

@ -93,7 +93,7 @@ def user_hotp():
@login_read_only
def user_otp_enable_self():
user_id = current_user.get_user_id()
r = ail_users.api_enable_user_otp(user_id, request.remote_addr)
r = ail_users.api_enable_user_otp(user_id, request.access_route[0])
if r[1] != 200:
return create_json_response(r[0], r[1])
current_user.kill_session()
@ -104,7 +104,7 @@ def user_otp_enable_self():
@login_read_only
def user_otp_disable_self():
user_id = current_user.get_user_id()
r = ail_users.api_disable_user_otp(user_id, request.remote_addr)
r = ail_users.api_disable_user_otp(user_id, request.access_route[0])
if r[1] != 200:
return create_json_response(r[0], r[1])
current_user.kill_session()
@ -115,7 +115,7 @@ def user_otp_disable_self():
@login_admin
def user_otp_reset_self(): # TODO ask for password ?
user_id = current_user.get_user_id()
r = ail_users.api_reset_user_otp(user_id, user_id, request.remote_addr)
r = ail_users.api_reset_user_otp(user_id, user_id, request.access_route[0])
if r[1] != 200:
return create_json_response(r[0], r[1])
else:
@ -127,7 +127,7 @@ def user_otp_reset_self(): # TODO ask for password ?
@login_admin
def user_otp_enable():
user_id = request.args.get('user_id')
r = ail_users.api_enable_user_otp(user_id, request.remote_addr)
r = ail_users.api_enable_user_otp(user_id, request.access_route[0])
if r[1] != 200:
return create_json_response(r[0], r[1])
user = ail_users.AILUser.get(user_id)
@ -139,7 +139,7 @@ def user_otp_enable():
@login_admin
def user_otp_disable():
user_id = request.args.get('user_id')
r = ail_users.api_disable_user_otp(user_id, request.remote_addr)
r = ail_users.api_disable_user_otp(user_id, request.access_route[0])
if r[1] != 200:
return create_json_response(r[0], r[1])
user = ail_users.AILUser.get(user_id)
@ -152,7 +152,7 @@ def user_otp_disable():
def user_otp_reset(): # TODO ask for password ?
user_id = request.args.get('user_id')
admin_id = current_user.get_user_id()
r = ail_users.api_reset_user_otp(admin_id, user_id, request.remote_addr)
r = ail_users.api_reset_user_otp(admin_id, user_id, request.access_route[0])
if r[1] != 200:
return create_json_response(r[0], r[1])
else:
@ -165,7 +165,7 @@ def user_otp_reset(): # TODO ask for password ?
@login_user
def new_token_user_self():
user_id = current_user.get_user_id()
r = ail_users.api_create_user_api_key_self(user_id, request.remote_addr)
r = ail_users.api_create_user_api_key_self(user_id, request.access_route[0])
if r[1] != 200:
return create_json_response(r[0], r[1])
else:
@ -177,7 +177,7 @@ def new_token_user_self():
def new_token_user():
user_id = request.args.get('user_id')
admin_id = current_user.get_user_id()
r = ail_users.api_create_user_api_key(user_id, admin_id, request.remote_addr)
r = ail_users.api_create_user_api_key(user_id, admin_id, request.access_route[0])
if r[1] != 200:
return create_json_response(r[0], r[1])
else:
@ -189,7 +189,7 @@ def new_token_user():
def user_logout():
user_id = request.args.get('user_id') # TODO LOGS
admin_id = current_user.get_user_id()
r = ail_users.api_logout_user(admin_id, user_id, request.remote_addr)
r = ail_users.api_logout_user(admin_id, user_id, request.access_route[0])
if r[1] != 200:
return create_json_response(r[0], r[1])
else:
@ -200,7 +200,7 @@ def user_logout():
@login_admin
def users_logout():
admin_id = current_user.get_user_id() # TODO LOGS
r = ail_users.api_logout_users(admin_id, request.remote_addr)
r = ail_users.api_logout_users(admin_id, request.access_route[0])
if r[1] != 200:
return create_json_response(r[0], r[1])
else:
@ -277,7 +277,7 @@ def create_user_post():
edit = True
else:
edit = False
ail_users.api_create_user(admin_id, request.remote_addr, email, password, org_uuid, role, enable_2_fa)
ail_users.api_create_user(admin_id, request.access_route[0], email, password, org_uuid, role, enable_2_fa)
new_user = {'email': email, 'password': str_password, 'org': org_uuid, 'otp': enable_2_fa, 'edited': edit}
return render_template("create_user.html", new_user=new_user, meta={}, all_roles=all_roles, acl_admin=True)
@ -294,7 +294,7 @@ def create_user_post():
def delete_user():
user_id = request.args.get('user_id')
admin_id = current_user.get_user_id()
r = ail_users.api_delete_user(user_id, admin_id, request.remote_addr)
r = ail_users.api_delete_user(user_id, admin_id, request.access_route[0])
if r[1] != 200:
return create_json_response(r[0], r[1])
else:
@ -334,7 +334,7 @@ def create_org_post():
name = request.form.get('name')
description = request.form.get('description')
r = ail_orgs.api_create_org(admin_id, org_uuid, name, request.remote_addr, description=description)
r = ail_orgs.api_create_org(admin_id, org_uuid, name, request.access_route[0], description=description)
if r[1] != 200:
return create_json_response(r[0], r[1])
else:
@ -350,7 +350,7 @@ def create_org_post():
def delete_org():
admin_id = current_user.get_user_id()
org_uuid = request.args.get('uuid')
r = ail_orgs.api_delete_org(org_uuid, admin_id, request.remote_addr)
r = ail_orgs.api_delete_org(org_uuid, admin_id, request.access_route[0])
if r[1] != 200:
return create_json_response(r[0], r[1])
else: