src/app.py hinzugefügt
This commit is contained in:
272
src/app.py
Normal file
272
src/app.py
Normal file
@ -0,0 +1,272 @@
|
|||||||
|
from flask import Flask, render_template, request, redirect, url_for, session, flash
|
||||||
|
from flask_socketio import SocketIO
|
||||||
|
from flask_sqlalchemy import SQLAlchemy
|
||||||
|
from werkzeug.security import generate_password_hash, check_password_hash
|
||||||
|
from terminal_backend import init_terminal_routes
|
||||||
|
from cryptography.fernet import Fernet
|
||||||
|
from flask import jsonify, send_from_directory
|
||||||
|
from werkzeug.utils import secure_filename
|
||||||
|
from dotenv import load_dotenv
|
||||||
|
import os
|
||||||
|
import uuid
|
||||||
|
import ipaddress
|
||||||
|
|
||||||
|
load_dotenv()
|
||||||
|
|
||||||
|
FERNET_SECRET = os.getenv('FERNET_SECRET')
|
||||||
|
SECRET_KEY = os.getenv('SECRET_KEY')
|
||||||
|
IP_ADDRESS = os.getenv('IP_ADDRESS')
|
||||||
|
PORT = os.getenv('PORT')
|
||||||
|
UPLOAD_FOLDER = os.path.join(os.path.dirname(__file__), 'sftp-cache')
|
||||||
|
|
||||||
|
os.makedirs(UPLOAD_FOLDER, exist_ok=True)
|
||||||
|
|
||||||
|
app = Flask(__name__)
|
||||||
|
app.secret_key = SECRET_KEY
|
||||||
|
|
||||||
|
fernet = Fernet(FERNET_SECRET.encode())
|
||||||
|
|
||||||
|
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///vaults.db'
|
||||||
|
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
|
||||||
|
app.config['UPLOAD_FOLDER'] = UPLOAD_FOLDER
|
||||||
|
|
||||||
|
db = SQLAlchemy(app)
|
||||||
|
socketio = SocketIO(app, manage_session=False)
|
||||||
|
|
||||||
|
user_vaults = db.Table(
|
||||||
|
'user_vaults',
|
||||||
|
db.Column('user_id', db.Integer, db.ForeignKey('local_user.id')),
|
||||||
|
db.Column('vault_id', db.Integer, db.ForeignKey('vault.id'))
|
||||||
|
)
|
||||||
|
|
||||||
|
class Vault(db.Model):
|
||||||
|
id = db.Column(db.String(36), primary_key=True)
|
||||||
|
name = db.Column(db.String(100), nullable=False)
|
||||||
|
host_ip = db.Column(db.String(100))
|
||||||
|
hostname = db.Column(db.String(100))
|
||||||
|
port = db.Column(db.Integer, default=22)
|
||||||
|
user = db.Column(db.String(100))
|
||||||
|
password_hash = db.Column(db.String(255))
|
||||||
|
ssh_key = db.Column(db.Text)
|
||||||
|
auth_type = db.Column(db.String(50))
|
||||||
|
|
||||||
|
class LocalUser(db.Model):
|
||||||
|
id = db.Column(db.Integer, primary_key=True)
|
||||||
|
username = db.Column(db.String(100), unique=True, nullable=False)
|
||||||
|
password_hash = db.Column(db.String(255), nullable=False)
|
||||||
|
password_changed = db.Column(db.Boolean, default=False)
|
||||||
|
role = db.Column(db.String(20), default='user')
|
||||||
|
vaults = db.relationship('Vault', secondary=user_vaults, backref='users')
|
||||||
|
|
||||||
|
|
||||||
|
@app.context_processor
|
||||||
|
def inject_user():
|
||||||
|
if 'user' in session:
|
||||||
|
current_user = LocalUser.query.filter_by(username=session['user']).first()
|
||||||
|
return {'current_user': current_user}
|
||||||
|
return {}
|
||||||
|
|
||||||
|
@app.route('/', methods=['GET', 'POST'])
|
||||||
|
def login():
|
||||||
|
if request.method == 'POST':
|
||||||
|
username = request.form['username']
|
||||||
|
password = request.form['password']
|
||||||
|
|
||||||
|
user = LocalUser.query.filter_by(username=username).first()
|
||||||
|
if user and check_password_hash(user.password_hash, password):
|
||||||
|
session['user'] = username
|
||||||
|
if not user.password_changed:
|
||||||
|
return redirect(url_for('change_password'))
|
||||||
|
return redirect(url_for('view_vaults'))
|
||||||
|
|
||||||
|
flash("Login fehlgeschlagen", "error")
|
||||||
|
return render_template('login.html')
|
||||||
|
|
||||||
|
@app.route('/change_password', methods=['GET', 'POST'])
|
||||||
|
def change_password():
|
||||||
|
if 'user' not in session:
|
||||||
|
return redirect(url_for('login'))
|
||||||
|
|
||||||
|
if request.method == 'POST':
|
||||||
|
new_password = request.form['new_password']
|
||||||
|
if new_password:
|
||||||
|
user = LocalUser.query.filter_by(username=session['user']).first()
|
||||||
|
user.password_hash = generate_password_hash(new_password)
|
||||||
|
user.password_changed = True
|
||||||
|
db.session.commit()
|
||||||
|
return redirect(url_for('view_vaults'))
|
||||||
|
|
||||||
|
return render_template('change_password.html')
|
||||||
|
|
||||||
|
@app.route('/vaults')
|
||||||
|
def view_vaults():
|
||||||
|
if 'user' not in session:
|
||||||
|
return redirect(url_for('login'))
|
||||||
|
|
||||||
|
current_user = LocalUser.query.filter_by(username=session['user']).first()
|
||||||
|
vaults = current_user.vaults
|
||||||
|
return render_template('dashboard.html', title="Hosts verwalten", vaults=vaults, page='vaults', current_user=current_user)
|
||||||
|
|
||||||
|
@app.route('/vaults/add', methods=['POST'])
|
||||||
|
def add_vault():
|
||||||
|
name = request.form.get('vault_name')
|
||||||
|
host_or_name = request.form.get('host_or_name')
|
||||||
|
port = request.form.get('port')
|
||||||
|
user_field = request.form.get('user')
|
||||||
|
password = request.form.get('password')
|
||||||
|
ssh_key = request.form.get('ssh_key')
|
||||||
|
auth_type = request.form.get('auth_type')
|
||||||
|
|
||||||
|
try:
|
||||||
|
ipaddress.ip_address(host_or_name)
|
||||||
|
host_ip = host_or_name
|
||||||
|
hostname = None
|
||||||
|
except ValueError:
|
||||||
|
host_ip = None
|
||||||
|
hostname = host_or_name
|
||||||
|
|
||||||
|
encrypted_pw = fernet.encrypt(password.encode()).decode() if password else None
|
||||||
|
|
||||||
|
if name:
|
||||||
|
new_vault = Vault(
|
||||||
|
id=str(uuid.uuid4()),
|
||||||
|
name=name,
|
||||||
|
host_ip=host_ip,
|
||||||
|
hostname=hostname,
|
||||||
|
port=int(port) if port else 22,
|
||||||
|
user=user_field,
|
||||||
|
password_hash=encrypted_pw,
|
||||||
|
ssh_key=ssh_key,
|
||||||
|
auth_type=auth_type
|
||||||
|
)
|
||||||
|
db.session.add(new_vault)
|
||||||
|
|
||||||
|
current_user = LocalUser.query.filter_by(username=session['user']).first()
|
||||||
|
current_user.vaults.append(new_vault)
|
||||||
|
|
||||||
|
db.session.commit()
|
||||||
|
return redirect(url_for('view_vaults'))
|
||||||
|
|
||||||
|
@app.route('/users', methods=['GET', 'POST'])
|
||||||
|
def users():
|
||||||
|
if 'user' not in session:
|
||||||
|
return redirect(url_for('login'))
|
||||||
|
|
||||||
|
current_user = LocalUser.query.filter_by(username=session['user']).first()
|
||||||
|
|
||||||
|
if request.method == 'POST':
|
||||||
|
if 'password' in request.form:
|
||||||
|
new_password = request.form.get('password')
|
||||||
|
if new_password:
|
||||||
|
current_user.password_hash = generate_password_hash(new_password)
|
||||||
|
current_user.password_changed = True
|
||||||
|
db.session.commit()
|
||||||
|
flash("Passwort wurde geändert.", "success")
|
||||||
|
|
||||||
|
elif 'new_username' in request.form:
|
||||||
|
new_username = request.form.get('new_username')
|
||||||
|
if new_username and new_username != current_user.username:
|
||||||
|
if LocalUser.query.filter_by(username=new_username).first():
|
||||||
|
flash("Benutzername bereits vergeben.", "error")
|
||||||
|
else:
|
||||||
|
current_user.username = new_username
|
||||||
|
session['user'] = new_username
|
||||||
|
db.session.commit()
|
||||||
|
flash("Benutzername geändert.", "success")
|
||||||
|
|
||||||
|
return render_template('profil.html', title="Profil", current_user=current_user, page='users')
|
||||||
|
|
||||||
|
@app.route('/sftp', methods=['GET'])
|
||||||
|
def sftp():
|
||||||
|
if 'user' not in session:
|
||||||
|
return redirect(url_for('login'))
|
||||||
|
current_user = LocalUser.query.filter_by(username=session['user']).first()
|
||||||
|
vaults = Vault.query.all() if current_user.role != 'user' else current_user.vaults
|
||||||
|
return render_template('sftp-modul.html', vaults=vaults)
|
||||||
|
|
||||||
|
@app.route('/users/delete/<username>', methods=['POST'])
|
||||||
|
def delete_user(username):
|
||||||
|
return redirect(url_for('view_vaults'))
|
||||||
|
|
||||||
|
@app.route('/vaults/delete/<vault_id>', methods=['POST'])
|
||||||
|
def delete_vault(vault_id):
|
||||||
|
vault = Vault.query.get(vault_id)
|
||||||
|
if vault:
|
||||||
|
db.session.delete(vault)
|
||||||
|
db.session.commit()
|
||||||
|
return redirect(url_for('view_vaults'))
|
||||||
|
|
||||||
|
@app.route('/vaults/edit/<vault_id>', methods=['POST'])
|
||||||
|
def edit_vault(vault_id):
|
||||||
|
vault = Vault.query.get(vault_id)
|
||||||
|
if vault:
|
||||||
|
vault.name = request.form.get("vault_name")
|
||||||
|
host_or_name = request.form.get("host_or_name")
|
||||||
|
try:
|
||||||
|
ipaddress.ip_address(host_or_name)
|
||||||
|
vault.host_ip = host_or_name
|
||||||
|
vault.hostname = None
|
||||||
|
except ValueError:
|
||||||
|
vault.host_ip = None
|
||||||
|
vault.hostname = host_or_name
|
||||||
|
|
||||||
|
vault.port = int(request.form.get("port")) if request.form.get("port") else 22
|
||||||
|
vault.user = request.form.get("user")
|
||||||
|
password = request.form.get("password")
|
||||||
|
if password:
|
||||||
|
vault.password_hash = generate_password_hash(password)
|
||||||
|
vault.ssh_key = request.form.get("ssh_key")
|
||||||
|
vault.auth_type = request.form.get("auth_type")
|
||||||
|
db.session.commit()
|
||||||
|
return redirect(url_for('view_vaults'))
|
||||||
|
|
||||||
|
@app.route('/sftp/upload', methods=['POST'])
|
||||||
|
def upload_file():
|
||||||
|
if 'file' not in request.files:
|
||||||
|
return jsonify({'error': 'Keine Datei hochgeladen'}), 400
|
||||||
|
file = request.files['file']
|
||||||
|
if file.filename == '':
|
||||||
|
return jsonify({'error': 'Leerer Dateiname'}), 400
|
||||||
|
filename = secure_filename(file.filename)
|
||||||
|
file.save(os.path.join(app.config['UPLOAD_FOLDER'], filename))
|
||||||
|
return jsonify({'success': True, 'filename': filename})
|
||||||
|
|
||||||
|
@app.route('/sftp/files', methods=['GET'])
|
||||||
|
def list_uploaded_files():
|
||||||
|
files = os.listdir(app.config['UPLOAD_FOLDER'])
|
||||||
|
return jsonify(files)
|
||||||
|
|
||||||
|
@app.route('/sftp/delete/<filename>', methods=['POST'])
|
||||||
|
def delete_uploaded_file(filename):
|
||||||
|
filepath = os.path.join(app.config['UPLOAD_FOLDER'], secure_filename(filename))
|
||||||
|
if os.path.exists(filepath):
|
||||||
|
os.remove(filepath)
|
||||||
|
return jsonify({'success': True})
|
||||||
|
return jsonify({'error': 'Datei nicht gefunden'}), 404
|
||||||
|
|
||||||
|
@app.route('/sftp/download/<filename>', methods=['GET'])
|
||||||
|
def download_uploaded_file(filename):
|
||||||
|
return send_from_directory(app.config['UPLOAD_FOLDER'], secure_filename(filename), as_attachment=True)
|
||||||
|
|
||||||
|
|
||||||
|
@app.route('/logout')
|
||||||
|
def logout():
|
||||||
|
session.clear()
|
||||||
|
return redirect(url_for('login'))
|
||||||
|
|
||||||
|
if __name__ == '__main__':
|
||||||
|
with app.app_context():
|
||||||
|
db.create_all()
|
||||||
|
if not LocalUser.query.filter_by(username='admin').first():
|
||||||
|
hashed_pw = generate_password_hash('changeme')
|
||||||
|
db.session.add(LocalUser(
|
||||||
|
username='admin',
|
||||||
|
password_hash=hashed_pw,
|
||||||
|
password_changed=False,
|
||||||
|
role='admin'
|
||||||
|
))
|
||||||
|
db.session.commit()
|
||||||
|
|
||||||
|
|
||||||
|
init_terminal_routes(app, socketio, db, Vault, LocalUser, fernet)
|
||||||
|
socketio.run(app, host=IP_ADDRESS, port=PORT, debug=True)
|
||||||
Reference in New Issue
Block a user