Hype-2/app.py

import os
import json
import requests
from flask import Flask , redirect , render_template, request, Response, stream_with_context, url_for, make_response, flash
from werkzeug.utils import secure_filename
from flask_sqlalchemy import SQLAlchemy
from flask_login import LoginManager , UserMixin , login_required ,login_user, logout_user,current_user
from werkzeug.security import generate_password_hash, check_password_hash

from config import *
from functions.fhost import *
from functions.flxc import *
from functions.fiso import *
from functions.fvm import *
from functions.fvnc import *
from functions.fbackup import *
from functions.fpool import *

app = Flask(__name__,static_url_path='',static_folder='static',template_folder='templates')

###########################
##       DROPZONE        ##
###########################
app.config['MAX_CONTENT_LENGTH'] = 10 * 1024 * 1024 * 1024
app.config['UPLOAD_EXTENSIONS'] = ['.iso']
app.config['UPLOAD_PATH'] = iso_path

############################
##         LOGIN          ##
############################

#Database (see base.sql for creation user table)
app.config['SQLALCHEMY_DATABASE_URI']='sqlite:///'+os.path.join(path,'db.db')
app.config['SECRET_KEY']='619619'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS']=True
db = SQLAlchemy(app)

#Login init (login_view => default login page)
login_manager = LoginManager()
login_manager.login_view = "/login"
login_manager.init_app(app)

#Database search
class User(UserMixin,db.Model):
    id = db.Column(db.Integer,primary_key=True,autoincrement=True)
    username = db.Column(db.String(200))
    email = db.Column(db.String(200))
    password = db.Column(db.String(200))

@login_manager.user_loader
def get(id):
    return User.query.get(id)

#Pass encrypt
def encrypt(password):
  hashed_pass=generate_password_hash(password)
  return hashed_pass

#Page for login
@app.route('/login',methods=['GET'])
def get_login():
    return render_template('login.html')

#Page for user creation (secured)
@app.route('/signup',methods=['GET'])
@login_required
def get_signup():
    return render_template('signup.html')

#Page for login, adding user and logout
@app.route('/login',methods=['POST'])
def login_post():
    email = request.form['email']
    password = request.form['password']
    user = User.query.filter_by(email=email).first()
    if user and check_password_hash(user.password, password):
      login_user(user)
      return redirect('/')
    return redirect('/')

@app.route('/signup',methods=['POST'])
@login_required
def signup_post():
    username = request.form['username']
    email = request.form['email']
    password = encrypt(request.form['password'])
    user = User(username=username,email=email,password=password)
    try:
        db.session.add(user)
        db.session.commit()
        flash(user+' added', category='success')
    except Exception as e:
        flash('Error on adding user :'+str(e), category='danger')
    return redirect('param')

@app.route('/logout',methods=['GET'])
def logout():
    logout_user()
    return redirect('/login')

############################
##         INDEX          ##
############################

@app.route('/',methods=['GET'])
@login_required
def get_home():
    lxc_up = len(get_lxc_activ())
    lxc_down = len(get_lxc_inactiv())
    vm_up = len(get_vm_activ())
    vm_down = len(get_vm_inactiv())
    full = get_host_full()
    return render_template('index.html', host=Host(), full = full, lxc_up=lxc_up, lxc_down=lxc_down, vm_up=vm_up, vm_down=vm_down)

def get_ressources():
    while True:
       full = get_host_full()
       json_data = json.dumps(
           {
              "time": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
              "cpu_percent": full['cpu']['percent'],
              "mem_percent": full['mem']['percent'],
              "swap_percent": full['swap']['percent'],
              "cpu_time_user": full['cpu']['time_user'],
              "cpu_time_nice": full['cpu']['time_nice'],
              "cpu_time_system": full['cpu']['time_system'],
              "cpu_time_idle": full['cpu']['time_idle'],
              "cpu_time_iowait": full['cpu']['time_iowait'],
              "cpu_time_irq": full['cpu']['time_irq'],
              "cpu_time_softirq": full['cpu']['time_softirq'],
              "cpu_time_steal": full['cpu']['time_steal'],
              "cpu_time_guest": full['cpu']['time_guest'],
              "cpu_time_guest_nice": full['cpu']['time_nice'],
              "mem_total": full['mem']['total'],
              "mem_available": full['mem']['available'],
              "mem_used": full['mem']['used'],
              "mem_free": full['mem']['free'],
              "mem_active": full['mem']['active'],
              "mem_inactive": full['mem']['inactive'],
              "mem_buffers": full['mem']['buffers'],
              "mem_cached": full['mem']['cached'],
              "mem_shared": full['mem']['shared'],
              "mem_slab": full['mem']['slab'],
              "swap_total": full['swap']['total'],
              "swap_used": full['swap']['used'],
              "swap_free": full['swap']['free'],
              "swap_sin": full['swap']['sin'],
              "swap_sout": full['swap']['sout'],
           }
       )
       yield f"data:{json_data}\n\n"
       time.sleep(1)

@app.route('/chart-ressources/')
def chart_data() -> Response:
    response = Response(stream_with_context(get_ressources()), mimetype="text/event-stream")
    response.headers["Cache-Control"] = "no-cache"
    response.headers["X-Accel-Buffering"] = "no"
    return response

############################
##         POOL           ##
############################

@app.route('/pool',methods=['GET'])
@login_required
def get_pool():
    return render_template('pool.html',list_pool_full = get_full_pool())

@app.route('/del_volume',methods=['POST'])
@login_required
def delvol():
    pool_name = request.form['pool_name']
    volume_name = request.form['volume_name']
    try:
        del_pool_vol(pool_name,volume_name)
        flash(volume_name+' deleted', category='success')
    except Exception as e:
        flash('Error on deleting '+volume_name+':'+str(e), category='danger')
    return redirect(url_for('get_pool'))

############################
##        VNC             ##
############################

####### TEST


####### ENDTEST

@app.route('/vnc',methods=['GET','POST',"DELETE"])
def vnc_connect():
    if request.method=='GET':
        resp = requests.get('http://localhost:6080/vnc.html')
        excluded_headers = ['content-encoding', 'content-length', 'transfer-encoding', 'connection']
        headers = [(name, value) for (name, value) in  resp.raw.headers.items() if name.lower() not in excluded_headers]
        response = Response(resp.content, resp.status_code, headers)
        return response
    elif request.method=='POST':
        resp = requests.post('http://localhost:6080/vnc.html',json=request.get_json())
        excluded_headers = ['content-encoding', 'content-length', 'transfer-encoding', 'connection']
        headers = [(name, value) for (name, value) in resp.raw.headers.items() if name.lower() not in excluded_headers]
        response = Response(resp.content, resp.status_code, headers)
        return response
    elif request.method=='DELETE':
        resp = requests.delete('http://localhost:6080/vnc.html').content
        response = Response(resp.content, resp.status_code, headers)
        return response

##TEST
@app.route('/ter',methods=['GET','POST',"DELETE"])
def ter_connect():
    if request.method=='GET':
        resp = requests.get('http://localhost:5008/')
        excluded_headers = ['content-encoding', 'content-length', 'transfer-encoding', 'connection']
        headers = [(name, value) for (name, value) in  resp.raw.headers.items() if name.lower() not in excluded_headers]
        response = Response(resp.content, resp.status_code, headers)
        return response
    elif request.method=='POST':
        resp = requests.post('http://localhost:5008/',json=request.get_json())
        excluded_headers = ['content-encoding', 'content-length', 'transfer-encoding', 'connection']
        headers = [(name, value) for (name, value) in resp.raw.headers.items() if name.lower() not in excluded_headers]
        response = Response(resp.content, resp.status_code, headers)
        return response
    elif request.method=='DELETE':
        resp = requests.delete('http://localhost:5008/').content
        response = Response(resp.content, resp.status_code, headers)
        return response


##ENDTEST
@app.route('/console_lxc',methods=['GET','POST','DELETE'])
@login_required
def consolelxc():
    lxc_name = request.form['lxc_name']
    try:
        pyxterm_connect(lxc_name)
        time.sleep(2)
        return render_template('terminal.html')
    except Exception as e:
        flash('Error starting Terminal :'+str(e), category='danger')
        return redirect(url_for('state'))

@app.route('/console_vm',methods=['POST'])
@login_required
def consolevm():
    vm_name = request.form['vm_name']
    try:
        socket_connect(vm_name)
        return render_template('vnc.html')
    except Exception as e:
        flash('Error starting VNC :'+str(e), category='danger')
        return redirect(url_for('state'))

############################
##       PARAMETERS       ##
############################
def get_users():
    users = User.query.all()
    list_users=[]
    for uni in users:
        single=[]
        single.append(uni.username)
        single.append(uni.email)
        list_users.append(single)
    return list_users


@app.route('/param',methods=['GET'])
@login_required
def param():
    return render_template('param.html',users_list = get_users())

@app.route('/deluser',methods=['POST'])
@login_required
def deluser():
    username = request.form['username']
    email = request.form['email']
    try:
        User.query.filter_by(username=username,email=email).delete()
        db.session.commit()
        flash(username+' deleted', category='success')
    except Exception as e:
        flash('Error on deleting '+username+':'+str(e), category='danger')
    return redirect('param')


############################
##         ISO           ##
############################

@app.route('/iso',methods=['GET'])
@login_required
def iso():
    return render_template('iso.html', list_iso = get_iso_list(), list_vm = get_vm_list(), list_iso_mount = get_iso_list())

@app.route('/deliso',methods=['POST'])
@login_required
def delete_iso():
    file=request.form['fichier']
    path=os.path.join(iso_path, file)
    try:
        os.remove(path)
        flash(file+' deleted', category='success')
    except Exception as e:
        flash('Error on deleting '+file+':'+str(e), category='danger')
    return render_template('iso.html', list_iso = get_iso_list(), list_vm = get_vm_list(), list_iso_mount = get_iso_list())

@app.route('/mountiso', methods=['POST'])
@login_required
def mountiso():
    vm_name=request.form['vm']
    if request.form['iso'] == '':
        iso_name = ''
    else:
        iso_name=iso_path+request.form['iso']
    try:
        mount_iso(vm_name,iso_name)
        flash(iso_name+' mounted', category='success')
    except Exception as e:
        flash('Error on mounting '+iso_name+':'+str(e), category='danger')
    return redirect(url_for('state'))

@app.route('/ejectiso', methods = ['POST'])
@login_required
def eject_iso():
    vm_name=request.form['ejectisovm']
    state = check_iso_is_mounted(vm_name)
    while state != 0:
        iso_file='';
        mount_iso(vm_name,iso_name)
        state = check_iso_is_mounted(vm_name)

@app.route('/iso', methods=['POST'])
@login_required
def upload_files():
    uploaded_file = request.files['file']
    filename = secure_filename(uploaded_file.filename)
    if filename != '':
        file_ext = os.path.splitext(filename)[1]
        if file_ext not in app.config['UPLOAD_EXTENSIONS']:
            return "Invalid image", 400
        uploaded_file.save(os.path.join(app.config['UPLOAD_PATH'], filename))
    return '', 204

############################
##         BACKUP         ##
############################

@app.route('/backup')
@login_required
def backup():
    full_snap_lxc=[]
    full_snap_vm=[]
    for lxc_name in get_lxc_list():
        if get_snap_list_lxc(lxc_name):
           full_snap_lxc.append(get_snap_list_lxc(lxc_name))
        else:
           full_snap_lxc.append('')
    for vm_name in get_vm_list():
        if get_snap_list_vm(vm_name):
           full_snap_vm.append(get_snap_list_vm(vm_name))
        else:
           full_snap_vm.append('')
    list_snap_lxc = zip(get_lxc_list(), full_snap_lxc)
    list_snap_vm = zip(get_vm_list(), full_snap_vm)
    return render_template('backup.html',list_snap_lxc=list_snap_lxc,list_snap_vm=list_snap_vm)

@app.route('/snaplxc', methods = ['POST'])
@login_required
def snaplxc():
   lxc_name=request.form['snap']
   try:
       create_snap_lxc(lxc_name)
       flash('Snapshot '+lxc_name+' done', category='success')
   except Exception as e:
       flash('Error on snapshoting '+lxc_name+':'+str(e), category='danger')
   return redirect(url_for('backup'))

@app.route('/del_snap_lxc', methods = ['POST'])
@login_required
def delsnaplxc():
    lxc_name=request.form['lxc_name']
    item=request.form['item']
    try:
        del_snap_lxc(lxc_name,item)
        flash(item+' deleted', category='success')
    except Exception as e:
        flash('Error on deleting '+item+':'+str(e), category='danger')
    return redirect(url_for('backup'))

@app.route('/rest_snap_lxc', methods = ['POST'])
@login_required
def restsnap_lxc():
    lxc_name=request.form['lxc_name']
    item=request.form['item']
    try:
        rest_snap_lxc(lxc_name,item)
        flash('Restore '+item+' done', category='success')
    except Exception as e:
        flash('Error restoring '+item+':'+str(e), category='danger')
    return redirect(url_for('backup'))


############################
##         BUILD          ##
############################

@app.route('/build',methods=['GET'])
@login_required
def build_lxc():
    return render_template('build.html', list_lxc_os=list_distrib(), list_profiles=get_os_profile_list(), list_net=conn.listNetworks(),list_iso = get_iso_list())

@app.route("/creation", methods=['POST'])
@login_required
def create_ct():
   lxc_ip = request.form['ip']
   try:
       create_lxc(request.form['nom'],request.form['os'])
       flash(request.form['nom']+' created', category='success')
   except Exception as e:
       flash('Error on creating '+request.form['nom']+':'+str(e), category='danger')
   if lxc_ip:
      set_lxc_ip(lxc_ip)
   return redirect(url_for('state'))

@app.route('/creationvm', methods=['POST'])
@login_required
def create_VM():
#VNC Version
    nom = request.form['vm_name']
    ram = request.form['ram']
    cpu = request.form['cpu']
    ose = request.form['os']
    iso = request.form['iso']
    iso = iso_path+iso
    net = request.form['net']
    disk = request.form['disk']
    creationcmd='--name '+str(nom)+' --ram '+str(ram)+' --disk pool=default,size='+str(disk)+',bus=virtio,format=qcow2 --vcpus '+str(cpu)+' --os-type linux --os-variant '+str(ose)+' --network network:bridged --graphics vnc,listen=0.0.0.0 --noautoconsole --console pty,target_type=serial --cdrom '+str(iso)+' --force --debug '
    try:
        os.system('virt-install '+creationcmd+' ')
        flash(str(nom)+' created', category='success')
    except Exception as e:
        flash('Error on creating '+str(nom)+':'+str(e),category='danger')
    return redirect(url_for('state'))

############################
##         STATE            ##
############################
@app.route('/state',methods=['GET'])
@login_required
def state():
    ips_lxc=[]
    for lxc_name in get_lxc_activ():
        ips_lxc.append(get_lxc_ip(lxc_name))
    activ_lxc=zip(get_lxc_activ(),ips_lxc)
    state_act_vm = []
    ips_vm = []
    for vm_name in get_vm_activ():
       state_act_vm.append(check_iso_is_mounted(vm_name))
       ips_vm.append(get_vm_ips(vm_name))
    activ_vm=zip(get_vm_activ(),state_act_vm,ips_vm)
    state_inact_vm = []
    for vm_name in get_vm_inactiv():
       state_inact_vm.append(check_iso_is_mounted(vm_name))
    inactiv_vm=zip(get_vm_inactiv(),state_inact_vm)
    return render_template('state.html', activ_vm=activ_vm, inactiv_vm=inactiv_vm,activ_lxc=activ_lxc,inactiv_lxc=get_lxc_inactiv())

############################
##         LXC            ##
############################
@app.route('/start_lxc',methods=['POST'])
@login_required
def startct():
    lxc_name = request.form['start']
    try:
        start_lxc(lxc_name)
        flash(lxc_name+' started', category='success')
    except Exception as e:
        flash('Error starting '+lxc_name+':'+str(e), category='danger')
    return redirect(url_for('state'))

@app.route('/stop_lxc',methods=['POST'])
@login_required
def stopct():
    lxc_name = request.form['stop']
    try:
        stop_lxc(lxc_name)
        flash(lxc_name+' stoped', category='success')
    except Exception as e:
        flash('Error stoping '+lxc_name+':'+str(e), category='danger')
    return redirect(url_for('state'))

@app.route('/destroy_lxc',methods=['POST'])
@login_required
def destroyct():
    lxc_name = request.form['destroy']
    try:
        destroy_lxc(lxc_name)
        flash(lxc_name+' destroyed', category='success')
    except Exception as e:
        flash('Error destroying '+lxc_name+':'+str(e), category='danger')
    return redirect(url_for('state'))

############################
##         VM             ##
############################
@app.route('/start_vm',methods=['POST'])
@login_required
def startvm():
    vm_name = request.form['start']
    try:
        start_vm(vm_name)
        flash(vm_name+' started', category='success')
    except Exception as e:
        flash('Error starting '+vm_name+':'+str(e), category='danger')
    return redirect(url_for('state'))

@app.route('/stop_vm',methods=['POST'])
@login_required
def stopvm():
    vm_name = request.form['stop']
    try:
        stop_vm(vm_name)
        flash(vm_name+' stoped', category='success')
    except Exception as e:
        flash('Error stoping '+vm_name+':'+str(e), category='danger')
    return redirect(url_for('state'))

@app.route('/destroy_vm',methods=['POST'])
@login_required
def destroyvm():
    vm_name = request.form['destroy']
    try:
        destroy_vm(vm_name)
        flash(vm_name+' destroyed', category='success')
    except Exception as e:
        flash('Error destroying '+vm_name+':'+str(e), category='danger')
    return redirect(url_for('state'))

############################
##         LOGIN          ##
############################
#Self certificate page
if __name__=='__main__':
    app.run(host=flask_config.host, port=flask_config.port, threaded=flask_config.thread, debug=flask_config.debug, ssl_context='adhoc')