Merge pull request #1 from CAELESTIS-Project-EU/security

Including security to the service

app.py

@@ -1,43 +1,114 @@
-from flask import Flask, request, jsonify
-import subprocess
-import json
+from flask import Flask, request, jsonify, flash, g
+from flask_sqlalchemy import SQLAlchemy
+from flask_httpauth import HTTPBasicAuth
+from flask_login import LoginManager, login_user, UserMixin
+from werkzeug.security import generate_password_hash, check_password_hash
+import jwt
+import time
+
+from config import service_conf
 
 app = Flask(__name__)
+app.config['SECRET_KEY'] = service_conf.secret_key
+app.config['SQLALCHEMY_DATABASE_URI'] = service_conf.database
+app.config['SQLALCHEMY_COMMIT_ON_TEARDOWN'] = True
+app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
 
+CAPTCHA_WEBSITE_KEY= service_conf.captcha_web_site_key
+CAPTCHA_SERVER_KEY= service_conf.captcha_site_key
 AVILABLE_SOFTWARE_JSON="config/available_software.json"
 
-def load_software():
-    with open(AVILABLE_SOFTWARE_JSON, 'r') as file:
-        return json.load(file)
+# db
+db = SQLAlchemy(app)
+
+#Basic Auth for API
+auth = HTTPBasicAuth()
+
+#Login for GUI
+login_manager = LoginManager()
+login_manager.login_view = 'auth.login'
+
+from contextlib import contextmanager
+
+@contextmanager
+def no_expire():
+    s = db.session()
+    s.expire_on_commit = False
+    try:
+        yield
+    finally:
+        s.expire_on_commit = True
+
+class User(UserMixin, db.Model):
+    id = db.Column(db.Integer, primary_key = True)
+    username = db.Column(db.String(32), index = True)
+    password_hash = db.Column(db.String(64))
+    email = db.Column(db.String(100))
+
+    def hash_password(self, password):
+        self.password_hash = generate_password_hash(password)
 
-@app.route('/run', methods=['POST'])
-def run_command():
+    def verify_password(self, password):
+        return check_password_hash(self.password_hash, password)
+
+    def generate_auth_token(self, expires_in=600):
+        return jwt.encode(
+            {'id': self.id, 'exp': time.time() + expires_in},
+            app.config['SECRET_KEY'], algorithm='HS256')
+
+@login_manager.user_loader
+def load_user(user_id):
+    return User.query.get(int(user_id))
+
+def check_and_log_user(username, password, remember):
+    user = User.query.filter_by(username=username).first()
+    if not user:
+        return False
+    if user.verify_password(password):
+        login_user(user, remember=remember)
+        return True 
+    else:
+        return False
+
+def verify_auth_token(token):
     try:
-        # Get the command from the request
-        data = request.get_json()
-        available_software = load_software()
-        software = data['software']
-        params = data['parameters']
-        command = available_software[software].format(**params)
-        # Run the command using subprocess
-        process = subprocess.Popen(command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
-        output, error = process.communicate()
-
-        # Check if the command executed successfully
-        if process.returncode == 0:
-            response = {'status': 'success', 'output': output.decode('utf-8')}
-        else:
-            response = {'status': 'failure', 'output': error.decode('utf-8')}
-        return jsonify(response), 200
-    except KeyError as e:
-        message = f"'software' or 'parameters' not found: {str(e)}"
-        return jsonify({'status': 'error', 'message': message}), 500
+        data = jwt.decode(token, app.config['SECRET_KEY'], algorithms=['HS256'])
+    except:
+        return
+    return User.query.get(data['id'])
+
+@auth.verify_password
+def verify_password(username_or_token, password):
+    # first try to authenticate by token
+    user = verify_auth_token(username_or_token)
+    if not user:
+        # try to authenticate with username/password
+        user = User.query.filter_by(username=username_or_token).first()
+        if not user:
+            flash("Username not found or token not valid")
+            return False       
+        if not user.verify_password(password):
+            flash("Password incorrect")
+            return False
+    g.user = user
+    return True
+
+def create_app():
+    login_manager.init_app(app)
+
+    from blueprints.auth import auth as auth_blueprint
+    app.register_blueprint(auth_blueprint)
+
+    from blueprints.dashboard import dashboard as dashboard_blueprint
+    app.register_blueprint(dashboard_blueprint)
+
+    from blueprints.api import api as api_blueprint
+    app.register_blueprint(api_blueprint)
 
-    except Exception as e:
-        message = f"Error: {type(e).__name__} - {str(e)}"
-        return jsonify({'status': 'error', 'message': message}), 500
+    return app
 
 
 if __name__ == '__main__':
+    app = create_app()
     app.run(debug=True)
 

blueprints/api.py

@@ -0,0 +1,71 @@
+from flask import Blueprint, request, jsonify
+import subprocess
+import json
+
+import importlib
+
+from app import AVILABLE_SOFTWARE_JSON, auth
+
+api = Blueprint('api', __name__)
+
+def load_software():
+    with open(AVILABLE_SOFTWARE_JSON, 'r') as file:
+        return json.load(file)
+
+def run_command(command):
+    # Run the command using subprocess
+    process = subprocess.Popen(command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+    output, error = process.communicate()
+
+    # Check if the command executed successfully
+    if process.returncode == 0:
+        response = {'status': 'success', 'output': output.decode('utf-8')}
+    else:
+        response = {'status': 'failure', 'output': error.decode('utf-8')}
+    return response
+
+def resolve_params(args, params):
+    for i in range(len(args)):
+        value = args[i]
+        if isinstance(value, str) and value[0] == '{' and value[-1] == '}':
+                variable_name = value[1:-1]
+                args[i] = params[variable_name]
+
+
+
+def run_function(full_func_name, args, params):
+        mod_name, func_name = full_func_name.rsplit('.',1)
+        module = importlib.import_module(mod_name)
+        resolve_params(args, params)
+        func = getattr(module, func_name)
+        return func(*args)
+
+
+@api.route('/run', methods=['POST'])
+@auth.login_required
+def run():
+    try:
+        # Get the command from the request
+        data = request.get_json()
+        print(str(data))
+        available_software = load_software()
+        software = data['software']
+        params = data['parameters']
+        s_type = available_software[software]['type']
+        if s_type == "command":
+            response = run_command(available_software[software]['command'].format(**params))
+        elif s_type == "function":
+            response = run_function(available_software[software]['function'], 
+                         available_software[software]['args'], params)
+        else :
+            raise Exception("Type of software " + software + "not supported (" + type + ")" )
+        return jsonify(response), 200
+    except KeyError as e:
+        message = f"'software' or 'parameters' not found: {str(e)}"
+        return jsonify({'status': 'error', 'message': message}), 500
+
+    except Exception as e:
+        import traceback
+        traceback.print_exc()
+        message = f"Error: {type(e).__name__} - {str(e)}"
+        return jsonify({'status': 'error', 'message': message}), 500

blueprints/auth.py

@@ -0,0 +1,71 @@
+from flask import Blueprint, render_template, redirect, url_for, request, flash
+from flask_login import login_required, logout_user
+import requests
+import json
+
+from app import db, User, check_and_log_user, CAPTCHA_SERVER_KEY, CAPTCHA_WEBSITE_KEY
+
+auth = Blueprint('auth', __name__)
+
+@auth.route('/login')
+def login():
+    return render_template('login.html')
+
+@auth.route('/login', methods=['POST'])
+def login_post():
+    username = request.form.get('username')
+    password = request.form.get('password')
+    remember = True if request.form.get('remember') else False
+    if check_and_log_user(username, password, remember):
+        return redirect(url_for('dashboard.home'))
+    else:
+        flash('Please check your login details and try again.')
+        return redirect(url_for('auth.login')) 
+
+
+
+@auth.route('/signup')
+def signup():
+    return render_template('signup.html', web_site_key=CAPTCHA_WEBSITE_KEY)
+
+@auth.route('/signup', methods=['POST'])
+def signup_post():
+    email = request.form.get('email')
+    username = request.form.get('username')
+    password = request.form.get('password')
+    retype_password = request.form.get('retype_password')
+    if CAPTCHA_WEBSITE_KEY:
+        captcha_response = request.form.get('g-recaptcha-response')
+        if captcha_response is None:
+            flash("Please, validate captcha")
+            return redirect(url_for('auth.signup'))
+        if not captcha_validation(captcha_response):
+            flash("Please, incorrect captcha validation")
+            return redirect(url_for('auth.signup'))
+    if username is None or password is None or retype_password is None or email is None:
+        flash('Email, username or password is not provided')
+        return redirect(url_for('auth.signup'))    
+    if User.query.filter_by(username=username).first() is not None:
+        flash('User already exists')
+        return redirect(url_for('auth.signup'), login=True)  
+    user = User(username=username, email=email)
+    user.hash_password(password)
+    db.session.add(user)
+    db.session.commit()
+    return redirect(url_for('auth.login'))
+
+def captcha_validation(captcha_response):
+    secret = CAPTCHA_SERVER_KEY
+    payload = {'response':captcha_response, 'secret':secret}
+    response = requests.post("https://www.google.com/recaptcha/api/siteverify", payload)
+    response_text = json.loads(response.text)
+    print("reCaptcha Response: " + str(response_text))
+    return response_text['success']
+
+
+@auth.route('/logout')
+@login_required
+def logout():
+    logout_user()
+    flash("Logout Successful")
+    return redirect(url_for('auth.login'))

blueprints/dashboard.py

@@ -0,0 +1,39 @@
+from flask import Blueprint, render_template, request, flash
+from flask_login import login_required, current_user
+from app import db
+
+dashboard = Blueprint('dashboard', __name__)
+
+@dashboard.route('/home')
+@login_required
+def home():
+    return render_template('main.html')
+
+@dashboard.route('/')
+@login_required
+def index():
+    return render_template('main.html')
+
+@dashboard.route('/account/token', methods=['POST'])
+@login_required
+def generate_token():
+    token = current_user.generate_auth_token()
+    return render_template('account.html', token=token)
+
+@dashboard.route('/account')
+@login_required
+def account():
+    return render_template('account.html', name=current_user.username)
+
+@dashboard.route('/account/update', methods=['POST'])
+@login_required
+def update_password():
+    new_password = request.form.get('new_password')
+    retype_password = request.form.get('retype_password')
+    if new_password == retype_password:
+        current_user.hash_password(new_password)
+        db.commit()
+    else:
+        flash("Password and retype are not the same")
+    return render_template('account.html', password=True)
+

config/available_software.json

@@ -1,3 +1,4 @@
 {
-	"list" : "ls {flags} {dir}"
+	"list" :  { "type" : "command", "command" : "ls {flags} {dir}" }, 
+	"test" : { "type" : "function", "function" : "functions.dummy.hello", "args" : ["{name}"] }
 }

config/service_conf.py

@@ -0,0 +1,4 @@
+secret_key = '_put_here_the_secret_key' #Put captcha google v2 web site key
+database = 'sqlite:///db.sqlite'
+captcha_web_site_key = None #To enable captcha put here your captcha google v2 web site key
+captcha_site_key= None #To enable captcha put here your captcha google v2 site key

functions/dummy.py

@@ -0,0 +1,3 @@
+def hello(name):
+    output = "hello " + name
+    return {'status': 'success', 'output': output}

requirements.txt

@@ -0,0 +1,6 @@
+Flask
+Flask-HTTPAuth
+Flask-Login
+Flask-SQLAlchemy==2.5.1 
+SQLAlchemy==1.4.32
+PyJWT

restore_db.py

@@ -0,0 +1,4 @@
+from app import db
+db.drop_all()
+db.create_all()
+exit()