import os import logging from datetime import datetime, timedelta from zoneinfo import ZoneInfo from dateutil.relativedelta import relativedelta from flask import Blueprint, render_template, request, redirect, session, url_for, flash, jsonify import config from auth.decorators import login_required from utils.audit import log_audit from utils.network import get_client_ip from utils.license import validate_license_key from db import get_connection, get_db_connection, get_db_cursor from models import get_licenses, get_license_by_id # Create Blueprint license_bp = Blueprint('licenses', __name__) @license_bp.route("/licenses") @login_required def licenses(): show_test = request.args.get('show_test', 'false') == 'true' licenses_list = get_licenses(show_test=show_test) return render_template("licenses.html", licenses=licenses_list, show_test=show_test) @license_bp.route("/license/edit/", methods=["GET", "POST"]) @login_required def edit_license(license_id): conn = get_connection() cur = conn.cursor() if request.method == "POST": try: # Get current license data for comparison current_license = get_license_by_id(license_id) if not current_license: flash('Lizenz nicht gefunden!', 'error') return redirect(url_for('licenses.licenses')) # Update license data new_values = { 'customer_name': request.form['customer_name'], 'customer_email': request.form['customer_email'], 'valid_from': request.form['valid_from'], 'valid_until': request.form['valid_until'], 'device_limit': int(request.form['device_limit']), 'active': 'active' in request.form } cur.execute(""" UPDATE licenses SET customer_name = %s, customer_email = %s, valid_from = %s, valid_until = %s, device_limit = %s, active = %s WHERE id = %s """, ( new_values['customer_name'], new_values['customer_email'], new_values['valid_from'], new_values['valid_until'], new_values['device_limit'], new_values['active'], license_id )) conn.commit() # Log changes log_audit('UPDATE', 'license', license_id, old_values={ 'customer_name': current_license['customer_name'], 'customer_email': current_license['customer_email'], 'valid_from': str(current_license['valid_from']), 'valid_until': str(current_license['valid_until']), 'device_limit': current_license['device_limit'], 'active': current_license['active'] }, new_values=new_values) flash('Lizenz erfolgreich aktualisiert!', 'success') # Preserve show_test parameter if present show_test = request.args.get('show_test', 'false') return redirect(url_for('licenses.licenses', show_test=show_test)) except Exception as e: conn.rollback() logging.error(f"Fehler beim Aktualisieren der Lizenz: {str(e)}") flash('Fehler beim Aktualisieren der Lizenz!', 'error') finally: cur.close() conn.close() # GET request license_data = get_license_by_id(license_id) if not license_data: flash('Lizenz nicht gefunden!', 'error') return redirect(url_for('licenses.licenses')) return render_template("edit_license.html", license=license_data) @license_bp.route("/license/delete/", methods=["POST"]) @login_required def delete_license(license_id): conn = get_connection() cur = conn.cursor() try: # Get license data before deletion license_data = get_license_by_id(license_id) if not license_data: flash('Lizenz nicht gefunden!', 'error') return redirect(url_for('licenses.licenses')) # Delete from sessions first cur.execute("DELETE FROM sessions WHERE license_key = %s", (license_data['license_key'],)) # Delete the license cur.execute("DELETE FROM licenses WHERE id = %s", (license_id,)) conn.commit() # Log deletion log_audit('DELETE', 'license', license_id, old_values={ 'license_key': license_data['license_key'], 'customer_name': license_data['customer_name'], 'customer_email': license_data['customer_email'] }) flash(f'Lizenz {license_data["license_key"]} erfolgreich gelöscht!', 'success') except Exception as e: conn.rollback() logging.error(f"Fehler beim Löschen der Lizenz: {str(e)}") flash('Fehler beim Löschen der Lizenz!', 'error') finally: cur.close() conn.close() # Preserve show_test parameter if present show_test = request.args.get('show_test', 'false') return redirect(url_for('licenses.licenses', show_test=show_test)) @license_bp.route("/create", methods=["GET", "POST"]) @login_required def create_license(): if request.method == "POST": customer_id = request.form.get("customer_id") license_key = request.form["license_key"].upper() # Immer Großbuchstaben license_type = request.form["license_type"] valid_from = request.form["valid_from"] is_test = request.form.get("is_test") == "on" # Checkbox value # Berechne valid_until basierend auf Laufzeit duration = int(request.form.get("duration", 1)) duration_type = request.form.get("duration_type", "years") start_date = datetime.strptime(valid_from, "%Y-%m-%d") if duration_type == "days": end_date = start_date + timedelta(days=duration) elif duration_type == "months": end_date = start_date + relativedelta(months=duration) else: # years end_date = start_date + relativedelta(years=duration) # Ein Tag abziehen, da der Starttag mitgezählt wird end_date = end_date - timedelta(days=1) valid_until = end_date.strftime("%Y-%m-%d") # Validiere License Key Format if not validate_license_key(license_key): flash('Ungültiges License Key Format! Erwartet: AF-YYYYMMFT-XXXX-YYYY-ZZZZ', 'error') return redirect(url_for('licenses.create_license')) # Resource counts domain_count = int(request.form.get("domain_count", 1)) ipv4_count = int(request.form.get("ipv4_count", 1)) phone_count = int(request.form.get("phone_count", 1)) device_limit = int(request.form.get("device_limit", 3)) conn = get_connection() cur = conn.cursor() try: # Prüfe ob neuer Kunde oder bestehender if customer_id == "new": # Neuer Kunde name = request.form.get("customer_name") email = request.form.get("email") if not name: flash('Kundenname ist erforderlich!', 'error') return redirect(url_for('licenses.create_license')) # Prüfe ob E-Mail bereits existiert if email: cur.execute("SELECT id, name FROM customers WHERE LOWER(email) = LOWER(%s)", (email,)) existing = cur.fetchone() if existing: flash(f'E-Mail bereits vergeben für Kunde: {existing[1]}', 'error') return redirect(url_for('licenses.create_license')) # Kunde einfügen (erbt Test-Status von Lizenz) cur.execute(""" INSERT INTO customers (name, email, is_test, created_at) VALUES (%s, %s, %s, NOW()) RETURNING id """, (name, email, is_test)) customer_id = cur.fetchone()[0] customer_info = {'name': name, 'email': email, 'is_test': is_test} # Audit-Log für neuen Kunden log_audit('CREATE', 'customer', customer_id, new_values={'name': name, 'email': email, 'is_test': is_test}) else: # Bestehender Kunde - hole Infos für Audit-Log cur.execute("SELECT name, email, is_test FROM customers WHERE id = %s", (customer_id,)) customer_data = cur.fetchone() if not customer_data: flash('Kunde nicht gefunden!', 'error') return redirect(url_for('licenses.create_license')) customer_info = {'name': customer_data[0], 'email': customer_data[1]} # Wenn Kunde Test-Kunde ist, Lizenz auch als Test markieren if customer_data[2]: # is_test des Kunden is_test = True # Lizenz hinzufügen cur.execute(""" INSERT INTO licenses (license_key, customer_id, license_type, valid_from, valid_until, is_active, domain_count, ipv4_count, phone_count, device_limit, is_test) VALUES (%s, %s, %s, %s, %s, TRUE, %s, %s, %s, %s, %s) RETURNING id """, (license_key, customer_id, license_type, valid_from, valid_until, domain_count, ipv4_count, phone_count, device_limit, is_test)) license_id = cur.fetchone()[0] # Ressourcen zuweisen try: # Prüfe Verfügbarkeit cur.execute(""" SELECT (SELECT COUNT(*) FROM resource_pools WHERE resource_type = 'domain' AND status = 'available' AND is_test = %s) as domains, (SELECT COUNT(*) FROM resource_pools WHERE resource_type = 'ipv4' AND status = 'available' AND is_test = %s) as ipv4s, (SELECT COUNT(*) FROM resource_pools WHERE resource_type = 'phone' AND status = 'available' AND is_test = %s) as phones """, (is_test, is_test, is_test)) available = cur.fetchone() if available[0] < domain_count: raise ValueError(f"Nicht genügend Domains verfügbar (benötigt: {domain_count}, verfügbar: {available[0]})") if available[1] < ipv4_count: raise ValueError(f"Nicht genügend IPv4-Adressen verfügbar (benötigt: {ipv4_count}, verfügbar: {available[1]})") if available[2] < phone_count: raise ValueError(f"Nicht genügend Telefonnummern verfügbar (benötigt: {phone_count}, verfügbar: {available[2]})") # Domains zuweisen if domain_count > 0: cur.execute(""" SELECT id FROM resource_pools WHERE resource_type = 'domain' AND status = 'available' AND is_test = %s LIMIT %s FOR UPDATE """, (is_test, domain_count)) for (resource_id,) in cur.fetchall(): cur.execute(""" UPDATE resource_pools SET status = 'allocated', allocated_to_license = %s, status_changed_at = CURRENT_TIMESTAMP, status_changed_by = %s WHERE id = %s """, (license_id, session['username'], resource_id)) cur.execute(""" INSERT INTO license_resources (license_id, resource_id, assigned_by) VALUES (%s, %s, %s) """, (license_id, resource_id, session['username'])) cur.execute(""" INSERT INTO resource_history (resource_id, license_id, action, action_by, ip_address) VALUES (%s, %s, 'allocated', %s, %s) """, (resource_id, license_id, session['username'], get_client_ip())) # IPv4s zuweisen if ipv4_count > 0: cur.execute(""" SELECT id FROM resource_pools WHERE resource_type = 'ipv4' AND status = 'available' AND is_test = %s LIMIT %s FOR UPDATE """, (is_test, ipv4_count)) for (resource_id,) in cur.fetchall(): cur.execute(""" UPDATE resource_pools SET status = 'allocated', allocated_to_license = %s, status_changed_at = CURRENT_TIMESTAMP, status_changed_by = %s WHERE id = %s """, (license_id, session['username'], resource_id)) cur.execute(""" INSERT INTO license_resources (license_id, resource_id, assigned_by) VALUES (%s, %s, %s) """, (license_id, resource_id, session['username'])) cur.execute(""" INSERT INTO resource_history (resource_id, license_id, action, action_by, ip_address) VALUES (%s, %s, 'allocated', %s, %s) """, (resource_id, license_id, session['username'], get_client_ip())) # Telefonnummern zuweisen if phone_count > 0: cur.execute(""" SELECT id FROM resource_pools WHERE resource_type = 'phone' AND status = 'available' AND is_test = %s LIMIT %s FOR UPDATE """, (is_test, phone_count)) for (resource_id,) in cur.fetchall(): cur.execute(""" UPDATE resource_pools SET status = 'allocated', allocated_to_license = %s, status_changed_at = CURRENT_TIMESTAMP, status_changed_by = %s WHERE id = %s """, (license_id, session['username'], resource_id)) cur.execute(""" INSERT INTO license_resources (license_id, resource_id, assigned_by) VALUES (%s, %s, %s) """, (license_id, resource_id, session['username'])) cur.execute(""" INSERT INTO resource_history (resource_id, license_id, action, action_by, ip_address) VALUES (%s, %s, 'allocated', %s, %s) """, (resource_id, license_id, session['username'], get_client_ip())) except ValueError as e: conn.rollback() flash(str(e), 'error') return redirect(url_for('licenses.create_license')) conn.commit() # Audit-Log log_audit('CREATE', 'license', license_id, new_values={ 'license_key': license_key, 'customer_name': customer_info['name'], 'customer_email': customer_info['email'], 'license_type': license_type, 'valid_from': valid_from, 'valid_until': valid_until, 'device_limit': device_limit, 'is_test': is_test }) flash(f'Lizenz {license_key} erfolgreich erstellt!', 'success') except Exception as e: conn.rollback() logging.error(f"Fehler beim Erstellen der Lizenz: {str(e)}") flash('Fehler beim Erstellen der Lizenz!', 'error') finally: cur.close() conn.close() # Preserve show_test parameter if present redirect_url = url_for('licenses.create_license') if request.args.get('show_test') == 'true': redirect_url += "?show_test=true" return redirect(redirect_url) # Unterstützung für vorausgewählten Kunden preselected_customer_id = request.args.get('customer_id', type=int) return render_template("index.html", username=session.get('username'), preselected_customer_id=preselected_customer_id)