pieter/eveAI
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor security to Flask-Security - Part 1

Browse Source
This commit is contained in:
Josako
2024-04-25 23:25:38 +02:00
parent dc235b5d2c
commit a37b551e53
15 changed files with 324 additions and 174 deletions
Download Patch File Download Diff File Expand all files Collapse all files

17
eveai_app/__init__.py
View File

@@ -1,12 +1,16 @@
import os
from flask import Flask
from .extensions import db, migrate, bcrypt, bootstrap, jwt
from .models.user import User, Tenant
from flask_security import SQLAlchemyUserDatastore
from werkzeug.middleware.proxy_fix import ProxyFix
from .extensions import db, migrate, bootstrap, security, mail, login_manager
from .models.user import User, Tenant, Role
from .models.document import Document, DocumentLanguage, DocumentVersion
def create_app(config_file=None):
app = Flask(__name__)
app.wsgi_app = ProxyFix(app.wsgi_app, x_for=1)
if config_file is None:
app.config.from_object('config.DevConfig')
@@ -19,6 +23,10 @@ def create_app(config_file=None):
pass
register_extensions(app)
# Setup Flask-Security-Too
user_datastore = SQLAlchemyUserDatastore(db, User, Role)
security.init_app(app, user_datastore)
register_blueprints(app)
print(app.config.get('SQLALCHEMY_DATABASE_URI'))
@@ -28,11 +36,12 @@ def create_app(config_file=None):
def register_extensions(app):
db.init_app(app)
migrate.init_app(app, db)
bcrypt.init_app(app)
bootstrap.init_app(app)
jwt.init_app(app)
mail.init_app(app)
login_manager.init_app(app)
# Register Blueprints
def register_blueprints(app):
from .views.user_views import user_bp
app.register_blueprint(user_bp)

11
eveai_app/extensions.py
View File

@@ -1,14 +1,17 @@
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate
from flask_bcrypt import Bcrypt
from flask_bootstrap import Bootstrap
from flask_jwt_extended import JWTManager
from flask_security import Security
from flask_mailman import Mail
from flask_login import LoginManager
# Create extensions
db = SQLAlchemy()
migrate = Migrate()
bcrypt = Bcrypt()
bootstrap = Bootstrap()
jwt = JWTManager()
security = Security()
mail = Mail()
login_manager = LoginManager()

34
eveai_app/models/user.py
View File

@@ -1,4 +1,5 @@
from ..extensions import db
from flask_security import UserMixin, RoleMixin
class Tenant(db.Model):
@@ -28,7 +29,24 @@ class Tenant(db.Model):
return '<Tenant %r>' % self.name
class User(db.Model):
class Role(db.Model, RoleMixin):
__bind_key__ = 'public'
__table_args__ = {'schema': 'public'}
id = db.Column(db.Integer(), primary_key=True)
name = db.Column(db.String(80), unique=True)
description = db.Column(db.String(255))
class RolesUsers(db.Model):
__bind_key__ = 'public'
__table_args__ = {'schema': 'public'}
user_id = db.Column(db.Integer(), db.ForeignKey('public.user.id', ondelete='CASCADE'), primary_key=True)
role_id = db.Column(db.Integer(), db.ForeignKey('public.role.id', ondelete='CASCADE'), primary_key=True)
class User(db.Model, UserMixin):
"""User model"""
__bind_key__ = 'public'
@@ -46,17 +64,19 @@ class User(db.Model):
first_name = db.Column(db.String(80), nullable=False)
last_name = db.Column(db.String(80), nullable=False)
is_active = db.Column(db.Boolean, default=True)
is_tester = db.Column(db.Boolean, default=False)
is_admin = db.Column(db.Boolean, default=False)
is_super = db.Column(db.Boolean, default=False)
fs_uniquifier = db.Column(db.String(255), unique=True, nullable=False)
confirmed_at = db.Column(db.DateTime, nullable=True)
valid_to = db.Column(db.Date, nullable=True)
# Login Information
last_login = db.Column(db.DateTime, nullable=True)
authenticated = db.Column(db.Boolean, default=False)
# Security Trackable Information
last_login_at = db.Column(db.DateTime, nullable=True)
current_login_at = db.Column(db.DateTime, nullable=True)
last_login_ip = db.Column(db.String(255), nullable=True)
current_login_ip = db.Column(db.String(255), nullable=True)
login_count = db.Column(db.Integer, nullable=False, default=0)
# Relations
roles = db.relationship('Role', secondary='public.roles_users', backref=db.backref('users', lazy='dynamic'))
tenant_id = db.Column(db.Integer, db.ForeignKey('public.tenant.id'), nullable=False)
def __repr__(self):

17
eveai_app/templates/login_user.html Normal file
View File

@@ -0,0 +1,17 @@
{% extends 'base.html' %}
{% block title %}Login{% endblock %}
{% block content %}
<form action="" method="post" novalidate>
<p>
{{ form.email.label }}<br>
{{ form.email(size=80) }}
</p>
<p>
{{ form.password.label }}<br>
{{ form.password(size=80) }}
</p>
<p>{{ form.submit() }}</p>
</form>
{% endblock %}

50
eveai_app/utils/security.py
View File

@@ -1,50 +0,0 @@
from functools import wraps
from flask_jwt_extended import get_jwt, verify_jwt_in_request
def super_required():
def wrapper(fn):
@wraps(fn)
def decorator(*args, **kwargs):
verify_jwt_in_request()
claims = get_jwt()
if not claims['is_super']:
return {'message': 'Authentication Error: Super users only!'}, 403
else:
return fn(*args, **kwargs)
return decorator
return wrapper
# Decorators
def admin_required():
def wrapper(fn):
@wraps(fn)
def decorator(*args, **kwargs):
verify_jwt_in_request()
claims = get_jwt()
if not claims['is_admin']:
return {'message': 'Authentication Error: Admins only!'}, 403
else:
return fn(*args, **kwargs)
return decorator
return wrapper
def tester_required():
def wrapper(fn):
@wraps(fn)
def decorator(*args, **kwargs):
verify_jwt_in_request()
claims = get_jwt()
if not claims['is_tester']:
return {'message': 'Authentication Error: Testers only!'}, 403
else:
return fn(*args, **kwargs)
return decorator
return wrapper

2
eveai_app/views/auth_forms.py
View File

@@ -6,5 +6,5 @@ from wtforms.validators import DataRequired, Length, Email
class LoginForm(FlaskForm):
email = EmailField('Email', validators=[DataRequired(), Email()])
password = PasswordField('Password', validators=[DataRequired(), Length(min=8)])
# remember_me = BooleanField('Remember me')
remember_me = BooleanField('Remember me')
submit = SubmitField('Login')

65
eveai_app/views/auth_views.py
View File

@@ -1,65 +1,46 @@
from datetime import datetime as dt, timezone as tz
from flask import request, redirect, url_for, flash, render_template, Blueprint, jsonify, session
from ..models.user import User, Tenant
from ..extensions import db, bcrypt
from .auth_forms import LoginForm
from flask_jwt_extended import (create_access_token, create_refresh_token, set_access_cookies, set_refresh_cookies,
unset_jwt_cookies)
from flask_security import login_user, logout_user
auth_bp = Blueprint('auth_bp', __name__)
from ..models.user import User, Tenant
from .auth_forms import LoginForm
auth_bp = Blueprint('auth_bp', __name__, template_folder='templates')
@auth_bp.route('/login', methods=['GET', 'POST'])
def login():
if request.method == 'POST':
email = request.form.get('email')
password = request.form.get('password')
# remember_me = True if request.form.get('remember_me') else False
form = LoginForm()
if form.validate_on_submit():
email = form.email.data
password = form.password.data
remember_me = True if form.remember_me.data else False
user = User.query.filter_by(email=email).first()
tenant = Tenant.query.filter_by(id=user.tenant_id).first()
if user:
if user and user.verify_and_update_password(password):
if user.is_active:
if bcrypt.check_password_hash(user.password, password):
response = jsonify({'msg': 'Login Successful'})
flash('Logged in successfully!', category='success')
login_user(user, remember=remember_me)
next_page = request.args.get('next')
# set session information
# session['user_id'] = user.id
# session['user_name'] = user.user_name
# session['email'] = user.email
# session['tenant_id'] = user.tenant_id
# session['tenant_name'] = tenant.name
session['tenant_id'] = user.tenant_id
session['tenant_name'] = tenant.name
# set JWT header information
additional_claims = {'tenant': user.tenant_id,
'is_super': user.is_super,
'is_admin': user.is_admin,
'is_tester': user.is_tester}
access_token = create_access_token(
identity=user.id,
additional_claims=additional_claims)
refresh_token = create_refresh_token(
identity=user.id,
additional_claims=additional_claims)
set_access_cookies(response, access_token)
set_refresh_cookies(response, refresh_token)
response.headers['Location'] = url_for('user_bp.user')
return response, 302
else:
flash('Incorrect email/password combination, try again.', category='error')
return redirect(next_page)
else:
flash('Account disabled. Please contact your administrator.', category='error')
else:
flash('Incorrect email/password combination, try again.', category='error')
flash('Invalid email or password.', category='error')
form = LoginForm()
return render_template('login.html', form=form)
@auth_bp.route('/logout', methods=['POST'])
def logout():
response = jsonify({'msg': 'Logout Successful'})
unset_jwt_cookies(response)
logout_user()
# Clear session data
session.pop('tenant_id', None)
session.pop('tenant_name', None)
return redirect(url_for('/'))

4
eveai_app/views/user_forms.py
View File

@@ -16,12 +16,10 @@ class UserForm(FlaskForm):
user_name = StringField('Name', validators=[DataRequired(), Length(max=80)])
email = EmailField('Email', validators=[DataRequired(), Email()])
password = PasswordField('Password', validators=[DataRequired(), Length(min=8)])
confirm_password = PasswordField('Confirm Password', validators=[DataRequired(), Length(min=8)])
first_name = StringField('First Name', validators=[DataRequired(), Length(max=80)])
last_name = StringField('Last Name', validators=[DataRequired(), Length(max=80)])
is_active = BooleanField('Is Active')
is_tester = BooleanField('Is Tester')
is_admin = BooleanField('Is Administrator')
is_super = BooleanField('Is Super User')
valid_to = DateField('Valid to', id='datepicker')
tenant_id = IntegerField('Tenant ID', validators=[NumberRange(min=0)])
submit = SubmitField('Submit')

95
eveai_app/views/user_views.py
View File

@@ -1,18 +1,17 @@
# from . import user_bp
from datetime import datetime as dt, timezone as tz
from flask import request, redirect, url_for, flash, render_template, Blueprint, session
from flask_jwt_extended import verify_jwt_in_request, get_jwt, get_jwt_identity, jwt_required
from flask_security import hash_password
from ..models.user import User, Tenant
from ..extensions import db, bcrypt
from ..extensions import db
from .user_forms import TenantForm, UserForm
from ..utils.database import Database
from ..utils.security import admin_required, super_required, tester_required
user_bp = Blueprint('user_bp', __name__, url_prefix='/user')
@user_bp.route('/tenant', methods=['GET', 'POST'])
@super_required
def tenant():
if request.method == 'POST':
# Handle the required attributes
@@ -65,61 +64,39 @@ def tenant():
@user_bp.route('/user', methods=['GET', 'POST'])
@admin_required
@jwt_required()
def user():
if request.method == 'POST':
# Handle the required attributes
username = request.form.get('user_name')
email = request.form.get('email')
password = request.form.get('password')
first_name = request.form.get('first_name')
last_name = request.form.get('last_name')
error = None
if not username:
error = 'Username is required.'
elif not email:
error = 'Email is required.'
elif not password:
error = 'Password is required.'
elif not first_name:
error = 'First name is required.'
elif not last_name:
error = 'Last name is required.'
if error is None:
password_hash = bcrypt.generate_password_hash(password).decode('utf-8')
# Create new user if there is no error
new_user = User(user_name=username, email=email, password=password_hash, first_name=first_name,
last_name=last_name)
# Handle optional attributes
new_user.is_active = bool(request.form.get('is_active'))
new_user.is_tester = bool(request.form.get('is_tester'))
new_user.is_admin = bool(request.form.get('is_admin'))
new_user.is_super = bool(request.form.get('is_super'))
new_user.valid_to = request.form.get('valid_to')
# Handle Timestamps
timestamp = dt.now(tz.utc)
new_user.created_at = timestamp
new_user.updated_at = timestamp
# Handle the relations
tenant_id = request.form.get('tenant_id')
the_tenant = Tenant.query.get(tenant_id)
new_user.tenant = the_tenant
# Add the new user to the database and commit the changes
try:
db.session.add(new_user)
db.session.commit()
except Exception as e:
error = e.args
flash(error) if error else flash('User added successfully.')
form = UserForm()
if form.validate_on_submit():
hashed_password = hash_password(form.password.data)
new_user = User(
user_name=form.user_name.data,
email=form.email.data,
password=hashed_password,
first_name=form.first_name.data,
last_name=form.last_name.data,
is_active=form.is_active.data,
valid_to=form.valid_to.data,
tenant_id=form.tenant_id.data
)
timestamp = dt.now(tz.utc)
new_user.created_at = timestamp
new_user.updated_at = timestamp
# Handle the relations
tenant_id = request.form.get('tenant_id')
the_tenant = Tenant.query.get(tenant_id)
new_user.tenant = the_tenant
# Add the new user to the database and commit the changes
try:
db.session.add(new_user)
db.session.commit()
flash('User added successfully.')
# return redirect(url_for('user/user'))
except Exception as e:
db.session.rollback()
flash(f'Failed to add user. Error: {str(e)}')
return render_template('user/user.html', form=form)