Files
proxmox_manager/create_database.py
2026-02-17 12:43:27 +01:00

243 lines
9.1 KiB
Python

"""
Script per creare il database di Proxmox Manager.
Uso:
python create_database.py
Requisiti:
- MySQL/MariaDB in esecuzione
- File .env configurato con DB_HOST, DB_USER, DB_PASSWORD, DB_NAME
- pip install -r requirements.txt
"""
import sys
import os
import bcrypt
import pymysql
from dotenv import load_dotenv
# Carica le variabili dal file .env
load_dotenv()
DB_HOST = os.getenv('DB_HOST', 'localhost')
DB_USER = os.getenv('DB_USER', 'root')
DB_PASSWORD = os.getenv('DB_PASSWORD', '')
DB_NAME = os.getenv('DB_NAME', 'proxmox_manager')
def hash_password(password: str) -> str:
"""Genera un hash bcrypt sicuro per la password."""
return bcrypt.hashpw(password.encode('utf-8'), bcrypt.gensalt(12)).decode('utf-8')
def get_connection(database=None):
"""Apre una connessione a MySQL."""
try:
conn = pymysql.connect(
host=DB_HOST,
user=DB_USER,
password=DB_PASSWORD,
database=database,
charset='utf8mb4',
cursorclass=pymysql.cursors.DictCursor,
autocommit=True,
)
return conn
except pymysql.Error as e:
print(f"\n[ERRORE] Impossibile connettersi a MySQL: {e}")
print("\nVerifica che:")
print(f" - MySQL/MariaDB sia in esecuzione su {DB_HOST}")
print(f" - Le credenziali nel file .env siano corrette (utente: {DB_USER})")
sys.exit(1)
def create_database(conn):
"""Crea il database se non esiste."""
with conn.cursor() as cur:
cur.execute(
f"CREATE DATABASE IF NOT EXISTS `{DB_NAME}` "
f"CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci"
)
print(f" [OK] Database '{DB_NAME}' pronto.")
def create_tables(conn):
"""Crea tutte le tabelle necessarie."""
tables = {
# ------------------------------------------------------------------ #
# 1. UTENTI #
# ------------------------------------------------------------------ #
'users': """
CREATE TABLE IF NOT EXISTS users (
id INT AUTO_INCREMENT PRIMARY KEY,
username VARCHAR(50) UNIQUE NOT NULL,
email VARCHAR(100) UNIQUE NOT NULL,
password_hash VARCHAR(255) NOT NULL,
is_admin BOOLEAN DEFAULT FALSE,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
last_login TIMESTAMP NULL,
active BOOLEAN DEFAULT TRUE,
INDEX idx_username (username),
INDEX idx_email (email)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
""",
# ------------------------------------------------------------------ #
# 2. ASSOCIAZIONE UTENTE ↔ VM #
# ------------------------------------------------------------------ #
'user_vms': """
CREATE TABLE IF NOT EXISTS user_vms (
id INT AUTO_INCREMENT PRIMARY KEY,
user_id INT NOT NULL,
vm_id INT NOT NULL,
vm_name VARCHAR(100),
notes TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE,
UNIQUE KEY unique_user_vm (user_id, vm_id),
INDEX idx_user_id (user_id),
INDEX idx_vm_id (vm_id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
""",
# ------------------------------------------------------------------ #
# 3. LOG AZIONI (audit trail) #
# ------------------------------------------------------------------ #
'action_logs': """
CREATE TABLE IF NOT EXISTS action_logs (
id INT AUTO_INCREMENT PRIMARY KEY,
user_id INT NOT NULL,
vm_id INT NOT NULL,
action_type ENUM(
'start','stop','restart','shutdown',
'backup','login','snapshot',
'subdomain_create','subdomain_delete','ip_assign'
) NOT NULL,
status ENUM('success','failed') NOT NULL,
error_message TEXT,
ip_address VARCHAR(45),
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE,
INDEX idx_user_id (user_id),
INDEX idx_vm_id (vm_id),
INDEX idx_created_at (created_at),
INDEX idx_action_type(action_type)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
""",
# ------------------------------------------------------------------ #
# 4. SOTTODOMINI CLOUDFLARE #
# ------------------------------------------------------------------ #
'subdomains': """
CREATE TABLE IF NOT EXISTS subdomains (
id INT AUTO_INCREMENT PRIMARY KEY,
user_id INT NOT NULL,
subdomain VARCHAR(100) NOT NULL,
ip_address VARCHAR(45) NOT NULL,
vm_id INT NULL,
proxied BOOLEAN DEFAULT FALSE,
cloudflare_record_id VARCHAR(100),
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE,
INDEX idx_user_id (user_id),
INDEX idx_subdomain(subdomain)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
""",
# ------------------------------------------------------------------ #
# 5. ASSEGNAZIONI IP (IPAM) #
# ------------------------------------------------------------------ #
'ip_assignments': """
CREATE TABLE IF NOT EXISTS ip_assignments (
id INT AUTO_INCREMENT PRIMARY KEY,
vm_id INT UNIQUE NOT NULL,
ip_address VARCHAR(45) NOT NULL,
mac_address VARCHAR(17) NULL,
notes TEXT NULL,
assigned_by INT NULL,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
FOREIGN KEY (assigned_by) REFERENCES users(id) ON DELETE SET NULL,
INDEX idx_vm_id (vm_id),
INDEX idx_ip_address(ip_address)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
""",
}
with conn.cursor() as cur:
for name, ddl in tables.items():
cur.execute(ddl)
print(f" [OK] Tabella '{name}' pronta.")
def create_default_users(conn):
"""Inserisce gli utenti di default (admin e testuser)."""
users = [
{
'username': 'admin',
'email': 'admin@localhost',
'password': 'admin123',
'is_admin': True,
},
{
'username': 'testuser',
'email': 'test@localhost',
'password': 'test123',
'is_admin': False,
},
]
with conn.cursor() as cur:
for u in users:
pw_hash = hash_password(u['password'])
cur.execute(
"""
INSERT INTO users (username, email, password_hash, is_admin)
VALUES (%s, %s, %s, %s)
ON DUPLICATE KEY UPDATE username = username
""",
(u['username'], u['email'], pw_hash, u['is_admin']),
)
print(f" [OK] Utente '{u['username']}' (password: {u['password']}) pronto.")
def main():
print("=" * 55)
print(" Proxmox Manager — Creazione Database")
print("=" * 55)
print(f"\nConnessione a MySQL ({DB_USER}@{DB_HOST})...")
# Prima connessione senza database (per crearlo)
conn_no_db = get_connection(database=None)
create_database(conn_no_db)
conn_no_db.close()
# Seconda connessione al database appena creato
conn = get_connection(database=DB_NAME)
print("\nCreazione tabelle...")
create_tables(conn)
print("\nCreazione utenti di default...")
create_default_users(conn)
conn.close()
print("\n" + "=" * 55)
print(" Database creato con successo!")
print("=" * 55)
print("\nCredenziali di accesso:")
print(" Amministratore : admin / admin123")
print(" Utente di test : testuser / test123")
print("\n IMPORTANTE: Cambia le password dopo il primo login!\n")
print("Per avviare l'applicazione:")
print(" python app.py")
print("\nPoi apri il browser su: http://localhost:5000\n")
if __name__ == '__main__':
main()