fix clean start

control_mqtt.py

@@ -1,4 +1,4 @@
-import paho.mqtt.subscribe as subscribe
+import paho.mqtt.client as mqtt
 import paho.mqtt.publish as publish
 import subprocess
 import argparse
@@ -8,6 +8,10 @@ import sys
 import os
 import logging
 
+# Configurazione Logging
+logging.basicConfig(level=logging.INFO, format='- PID: %(process)d %(levelname)8s: %(message)s')
+logger = logging.getLogger(__name__)
+
 class CurrentClients:
     def __init__(self, args):
         with open(args.dyn_sec_conf, "r") as file:
@@ -17,8 +21,11 @@ class CurrentClients:
         self.venv_path = sys.prefix
         for username in self.active_clients_list:
             if username.endswith("_ase"):
-                self.start_client(username, args)
-                logging.info(f"Init start client for {username}")
+                try:
+                    self.start_client(username, args)
+                    logger.info(f"Init start client for {username}")
+                except Exception as e:
+                    logger.error(f"Error starting client {username}: {e}")
 
     def list(self):
         return self.active_clients_list
@@ -28,77 +35,117 @@ class CurrentClients:
 
     def add(self, client, args):
         self.active_clients_list.append(client)
-        self.start_client(client, args) if client.endswith("_ase") else None
+        try:
+            self.start_client(client, args) if client.endswith("_ase") else None
+        except Exception as e:
+            logger.error(f"Error adding client {client}: {e}")
 
     def remove(self, client):
         self.active_clients_list.remove(client)
-        self.stop_client(client) if client.endswith("_ase") else None
+        try:
+            self.stop_client(client) if client.endswith("_ase") else None
+        except Exception as e:
+            logger.error(f"Error removing client {client}: {e}")
 
     def start_client(self, client, args):
-        process = subprocess.Popen(
-            [f'{self.venv_path}/bin/python3', args.ase_receiver, f'--client={client}'],
-            stdout=subprocess.PIPE,
-            stderr=subprocess.PIPE,
-            text=True
-        )
-        '''
-        for line in process.stdout:
-            logging.info(f"Subtask stdout: {line.strip()}")
-
-        for line in process.stderr:
-            logging.error(f"Subtask stderr: {line.strip()}")
-        '''
-        self.active_clients_pids[client] = process.pid
-        logging.info(f"Started process for {client}, PID: {process.pid}")
+        try:
+            process = subprocess.Popen(
+                [f'{self.venv_path}/bin/python3', args.ase_receiver, f'--client={client}'],
+                stdout=subprocess.PIPE,
+                stderr=subprocess.PIPE,
+                text=True
+            )
+            self.active_clients_pids[client] = process.pid
+            logger.info(f"Started process for {client}, PID: {process.pid}")
+        except Exception as e:
+            logger.error(f"Error starting process for {client}: {e}")
 
     def stop_client(self, client):
-        logging.info(f"Terminating process for {client} (PID: {self.active_clients_pids[client]})")
-        os.kill(self.active_clients_pids[client], 9)
+        try:
+            logger.info(f"Terminating process for {client} (PID: {self.active_clients_pids[client]})")
+            os.kill(self.active_clients_pids[client], 9)
+        except Exception as e:
+            logger.error(f"Error stopping client {client}: {e}")
 
 def get_client_list(args, auth):
-    publish.single(args.pub_topic, '{"commands":[{"command":"listClients"}]}', hostname=args.host, port=args.port, auth=auth)
+    try:
+        publish.single(args.pub_topic, '{"commands":[{"command":"listClients"}]}', hostname=args.host, port=args.port, auth=auth)
+    except Exception as e:
+        logger.error(f"Error publishing client list request: {e}")
 
 def create_client(datas, userdata):
-    get_client_list(userdata['args'], userdata['auth'])
+    try:
+        get_client_list(userdata['args'], userdata['auth'])
+    except Exception as e:
+        logger.error(f"Error creating client: {e}")
 
 def delete_client(datas, userdata):
-    get_client_list(userdata['args'], userdata['auth'])
+    try:
+        get_client_list(userdata['args'], userdata['auth'])
+    except Exception as e:
+        logger.error(f"Error deleting client: {e}")
 
 def list_clients(datas, userdata):
-    list_clients = datas['responses'][0]['data']['clients']
+    try:
+        list_clients = datas['responses'][0]['data']['clients']
 
-    delta_clients_add = set(list_clients) - set(userdata['cur_clients'].list())
-    [userdata['cur_clients'].add(item, userdata['args']) for item in delta_clients_add]
+        delta_clients_add = set(list_clients) - set(userdata['cur_clients'].list())
+        [userdata['cur_clients'].add(item, userdata['args']) for item in delta_clients_add]
 
-    delta_clients_del = set(userdata['cur_clients'].list()) - set(list_clients)
-    [userdata['cur_clients'].remove(item) for item in delta_clients_del]
+        delta_clients_del = set(userdata['cur_clients'].list()) - set(list_clients)
+        [userdata['cur_clients'].remove(item) for item in delta_clients_del]
+    except Exception as e:
+        logger.error(f"Error listing clients: {e}")
 
 def ctrl_client_mod(client, userdata, message):
+    try:
+        command_functions = {
+            "createClient": create_client,
+            "deleteClient": delete_client,
+            "listClients": list_clients,
+        }
 
-    command_functions = {
-        "createClient": create_client,
-        "deleteClient": delete_client,
-        "listClients": list_clients,
-    }
-
-    datas = json.loads(message.payload)
-    target_commands = {"createClient", "deleteClient", "listClients"}
-    found_command = [item["command"] for item in datas['responses'] if item["command"] in target_commands]
-    if found_command:
-        command_functions[found_command[0]](datas, userdata)
+        datas = json.loads(message.payload.decode('utf-8'))
+        target_commands = {"createClient", "deleteClient", "listClients"}
+        found_command = [item["command"] for item in datas['responses'] if item["command"] in target_commands]
+        if found_command:
+            command_functions[found_command[0]](datas, userdata)
+    except Exception as e:
+        logger.error(f"Error processing client command: {e}")
 
 def get_credentials(args):
-    url = args.wallet + "get"
-    data = {
-        "master_password":  os.getenv('WALLET_MASTER_PASSWORD'),
-        "site": "mqtt_control"
-    }
-    response = requests.post(url, json=data)
-    if response.status_code != 200:
-        logging.error(f"Error to get pwd from wallet.")
+    try:
+        url = args.wallet + "get"
+        data = {
+            "master_password":  os.getenv('WALLET_MASTER_PASSWORD'),
+            "site": "mqtt_control"
+        }
+        response = requests.post(url, json=data)
+        if response.status_code != 200:
+            logger.error("Error to get pwd from wallet.")
+            exit(1)
+
+        return response.json().get('password')
+    except Exception as e:
+        logger.error(f"Error getting credentials: {e}")
         exit(1)
 
-    return response.json().get('password')
+def on_connect(client, userdata, flags, rc, properties=None):
+    try:
+        if rc == 0:
+            logger.info("Connected successfully")
+            client.subscribe(userdata['args'].sub_topic, userdata['args'].qos)
+        else:
+            logger.error(f"Connection failed with code {rc}")
+    except Exception as e:
+        logger.error(f"Error on connect: {e}")
+
+def on_message(client, userdata, message):
+    try:
+        #logger.info(f"Received message on {message.topic}: {message.payload}")
+        ctrl_client_mod(client, userdata, message)
+    except Exception as e:
+        logger.error(f"Error handling message: {e}")
 
 def main():
     parser = argparse.ArgumentParser()
@@ -106,36 +153,38 @@ def main():
     parser.add_argument('-H', '--host', default="mqtt")
     parser.add_argument('-t', '--sub_topic', default="$CONTROL/dynamic-security/v1/response")
     parser.add_argument('-T', '--pub_topic', default="$CONTROL/dynamic-security/v1")
-    parser.add_argument('-q', '--qos', type=int,default=1)
+    parser.add_argument('-q', '--qos', type=int, default=1)
     parser.add_argument('-u', '--username', default="admin")
     parser.add_argument('-w', '--wallet', default="http://localhost:5000/")
     parser.add_argument('-P', '--port', type=int, default=1883)
-    parser.add_argument('-L', '--log_level', choices=['DEBUG', 'INFO', 'WARNING', 'ERROR'], default='INFO')
     parser.add_argument('-s', '--dyn_sec_conf', default='./dynamic-security.json')
     parser.add_argument('-r', '--ase_receiver', default='./subscribe_ase_receiver.py')
 
     args = parser.parse_args()
 
-    logging.basicConfig(
-        format="- PID: %(process)d %(levelname)8s: %(message)s ",
-        level=args.log_level
-    )
-
-
-    auth = {'username': args.username, 'password': get_credentials(args)}
-    cur_clients = CurrentClients(args)
-    userdata = {'args': args, 'cur_clients': cur_clients, 'auth': auth}
-
     try:
-        subscribe.callback(ctrl_client_mod, hostname=args.host, port=args.port,
-                           topics=args.sub_topic,
-                           auth=auth, userdata=userdata)
+        auth = {'username': args.username, 'password': get_credentials(args)}
+        cur_clients = CurrentClients(args)
+        userdata = {'args': args, 'cur_clients': cur_clients, 'auth': auth}
+
+        client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, protocol=mqtt.MQTTv5)
+        client.username_pw_set(auth['username'], auth['password'])
+
+        client.on_connect = on_connect
+        client.on_message = on_message
+
+        client.user_data_set(userdata)
+
+        client.connect(args.host, args.port)
+        client.loop_forever()
     except (KeyboardInterrupt, Exception) as e:
-        logging.info("Terminating: ....")
-
-        for client in cur_clients.list():
-            cur_clients.stop_client(client) if client.endswith("_ase") else None
-
+        logger.info("Terminating: ....")
+        logger.error(f"Error in main loop: {e}")
+        for client_name in cur_clients.list():
+            try:
+                cur_clients.stop_client(client_name) if client_name.endswith("_ase") else None
+            except Exception as stop_e:
+                logger.error(f"Error stopping client {client_name}: {stop_e}")
 
 if __name__ == "__main__":
-    main()
+    main()

password_wallet_api.py

@@ -46,7 +46,11 @@ def init_db():
             id SERIAL PRIMARY KEY,
             site TEXT NOT NULL,
             username TEXT NOT NULL,
-            password TEXT NOT NULL
+            password TEXT NOT NULL,
+            client_id TEXT NOT NULL,
+            topic TEXT NOT NULL,
+            created_at timestamptz DEFAULT CURRENT_TIMESTAMP,
+            CONSTRAINT site_user_clientid_unique UNIQUE(site, username, client_id)
         )
     """)
     conn.commit()
@@ -89,14 +93,14 @@ def authenticate(master_password):
     return auth_success
 
 # Aggiungi una password al database
-def add_password(site, username, password, cipher):
+def add_password(site, username, password, client_id, topic, cipher):
     conn = get_db_connection()
     cursor = conn.cursor()
     encrypted_password = cipher.encrypt(password.encode()).decode()
     try:
         cursor.execute(
-            f"INSERT INTO {DB_CONFIG['dbschema']}.{DB_CONFIG['dbtable']} (site, username, password) VALUES (%s, %s, %s)",
-            (site, username, encrypted_password))
+            f"INSERT INTO {DB_CONFIG['dbschema']}.{DB_CONFIG['dbtable']} (site, username, password, client_id, topic) VALUES (%s, %s, %s, %s, %s)",
+            (site, username, encrypted_password, client_id, topic))
         conn.commit()
         logging.info(f"Password aggiunta per il sito: {site}.")
     except psycopg2.Error as e:
@@ -109,18 +113,18 @@ def get_password(site, cipher):
     conn = get_db_connection()
     cursor = conn.cursor()
     try:
-        cursor.execute(f"SELECT username, password FROM {DB_CONFIG['dbschema']}.{DB_CONFIG['dbtable']} WHERE site = %s", (site,))
+        cursor.execute(f"SELECT username, password, client_id, topic FROM {DB_CONFIG['dbschema']}.{DB_CONFIG['dbtable']} WHERE site = %s", (site,))
         row = cursor.fetchone()
         if row:
-            username, encrypted_password = row
+            username, encrypted_password, client_id, topic = row
             decrypted_password = cipher.decrypt(encrypted_password.encode()).decode()
             logging.info(f"Password recuperata per il sito: {site}.")
-            return username, decrypted_password
+            return username, decrypted_password, client_id, topic
         logging.warning(f"Sito non trovato: {site}.")
-        return None, None
+        return None, None, None, None
     except psycopg2.Error as e:
         logging.error(f"Errore durante il recupero della password: {e}")
-        return None, None
+        return None, None, None, None
     finally:
         conn.close()
 
@@ -162,6 +166,8 @@ def add_password_api():
     site = request.json.get('site')
     username = request.json.get('username')
     password = request.json.get('password')
+    client_id = request.json.get('client_id')
+    topic = request.json.get('topic')
 
     if not authenticate(master_password):
         logging.warning("Tentativo di aggiungere una password con master password errata.")
@@ -169,7 +175,7 @@ def add_password_api():
 
     key = derive_key(master_password)
     cipher = Fernet(key)
-    add_password(site, username, password, cipher)
+    add_password(site, username, password, client_id, topic, cipher)
     return jsonify({"message": "Password aggiunta con successo"})
 
 # Endpoint per recuperare una password
@@ -184,12 +190,12 @@ def get_password_api():
 
     key = derive_key(master_password)
     cipher = Fernet(key)
-    username, password = get_password(site, cipher)
+    username, password, client_id, topic = get_password(site, cipher)
 
     if username is None:
         return jsonify({"error": "Sito non trovato"}), 404
 
-    return jsonify({"site": site, "username": username, "password": password})
+    return jsonify({"site": site, "username": username, "password": password, "client_id": client_id, "topic": topic})
 
 # Endpoint per cancellare una password
 @app.route('/delete', methods=['POST'])

subscribe_ase_receiver.py

@@ -1,12 +1,17 @@
-import paho.mqtt.subscribe as subscribe
+import paho.mqtt.client as mqtt
+from paho.mqtt.properties import Properties
+from paho.mqtt.packettypes import PacketTypes
 import argparse
 import requests
 import psycopg2
 import json
-import sys
 import os
 import logging
 
+# Configurazione Logging
+logging.basicConfig(level=logging.INFO, format='%(asctime)s - PID: %(process)d %(levelname)8s: %(message)s', filename="/var/log/ase_receiver.log")
+logger = logging.getLogger()
+
 # Configurazione connessione PostgreSQL
 DB_CONFIG = {
     "dbname": os.getenv("DB_NAME"),
@@ -21,15 +26,15 @@ DB_CONFIG = {
 def get_credentials(args):
     url = args.wallet + "get"
     data = {
-        "master_password":  os.getenv('WALLET_MASTER_PASSWORD'),
+        "master_password": os.getenv('WALLET_MASTER_PASSWORD'),
         "site": f"{args.client}_site"
     }
     response = requests.post(url, json=data)
     if response.status_code != 200:
-        logging.error(f"Error to get pwd from wallet.")
+        logger.error("Error to get pwd from wallet.")
         exit(1)
 
-    return response.json().get('password')
+    return response.json().get('password'), response.json().get('client_id'), response.json().get('topic')
 
 def get_db_connection():
     return psycopg2.connect(
@@ -40,8 +45,7 @@ def get_db_connection():
         port=DB_CONFIG["port"]
     )
 
-# Inizializza il database
-def init_db(args):
+def init_db(args, main_topic):
     try:
         conn = get_db_connection()
         cursor = conn.cursor()
@@ -50,7 +54,8 @@ def init_db(args):
                 id bigserial NOT NULL,
                 main_topic text NOT NULL,
                 tt_data jsonb NULL,
-                created_at timestamp DEFAULT CURRENT_TIMESTAMP NULL
+                created_at timestamp DEFAULT CURRENT_TIMESTAMP NULL,
+                CONSTRAINT {DB_CONFIG['dbtable']}_pkey PRIMARY KEY (id, main_topic)
             )
             PARTITION BY LIST (main_topic);
         """)
@@ -61,27 +66,34 @@ def init_db(args):
         """)
         conn.commit()
         cursor.execute(f"""
-            CREATE TABLE IF NOT EXISTS {DB_CONFIG['dbschema']}.{DB_CONFIG['dbtable']}_{args.client.removesuffix("_ase")} PARTITION OF {DB_CONFIG['dbschema']}.{DB_CONFIG['dbtable']}
-                FOR VALUES IN ('{args.client.removesuffix("_ase")}')
+            CREATE TABLE IF NOT EXISTS {DB_CONFIG['dbschema']}.{DB_CONFIG['dbtable']}_{main_topic} PARTITION OF {DB_CONFIG['dbschema']}.{DB_CONFIG['dbtable']}
+                FOR VALUES IN ('{main_topic}')
         """)
         conn.commit()
     except Exception as e:
-        logging.error(f"Errore durante l'inizializzazione del database: {e}")
+        logger.error(f"Errore durante l'inizializzazione del database: {e}")
         exit(1)
     finally:
         conn.close()
-    logging.info("Database inizializzato.")
+    logger.info("Database inizializzato.")
 
 def create_nested_json(path, data):
-    keys = path.split('/')[1:]
+    main_topic =  path.split('/')[0]
+    keys =  path.split('/')[1:]
     nested_json = data
     for key in reversed(keys):
         nested_json = {key: nested_json}
-    return nested_json
+    return main_topic, nested_json
 
-def receive_data(client, userdata, message):
+def on_connect(client, userdata, flags, rc, properties):
+    if rc == 0:
+        logger.info("Connesso al broker MQTT")
+    else:
+        logger.error(f"Errore di connessione, codice: {rc}")
+
+def on_message(client, userdata, message):
     datastore = json.loads(message.payload)
-    json_data = create_nested_json(message.topic, datastore)
+    main_topic, json_data = create_nested_json(message.topic, datastore)
 
     try:
         conn = get_db_connection()
@@ -90,45 +102,65 @@ def receive_data(client, userdata, message):
             INSERT INTO {DB_CONFIG['dbschema']}.{DB_CONFIG['dbtable']}
                 (main_topic, tt_data)
             VALUES
-                ('{userdata['args'].client.removesuffix("_ase")}', '{json.dumps(json_data)}'::jsonb);
+                ('{main_topic}', '{json.dumps(json_data)}'::jsonb);
         """)
         conn.commit()
     except Exception as e:
-        logging.error(f"Errore durante l'inserimento dei dati nel database: {e}")
+        logger.error(f"Errore durante l'inserimento dei dati nel database: {e}")
     finally:
         conn.close()
 
+def on_disconnect(client, userdata, rc, properties=None):
+    if rc != 0:
+        logger.warning(f"Disconnesso dal broker con codice {rc}. Riconnessione...")
+        client.reconnect()
 
 def main():
     parser = argparse.ArgumentParser()
 
     parser.add_argument('-H', '--host', default="mqtt")
-    parser.add_argument('-q', '--qos', type=int,default=2)
+    parser.add_argument('-q', '--qos', type=int, default=1)
     parser.add_argument('-P', '--port', type=int, default=1883)
     parser.add_argument('-c', '--client')
     parser.add_argument('-w', '--wallet', default="http://mqtt:5000/")
-    parser.add_argument('-L', '--log_level', choices=['DEBUG', 'INFO', 'WARNING', 'ERROR'], default='INFO')
 
     args = parser.parse_args()
 
-    logging.basicConfig(
-        format="- PID: %(process)d %(levelname)8s: %(message)s ",
-        level=args.log_level,
-        stream=sys.stderr
-    )
-    init_db(args)
 
-    auth = {'username': args.client, 'password': get_credentials(args)}
+    password, client_id, topic = get_credentials(args)
+
+    main_topic = topic.split('/')[0]
+
+    init_db(args, main_topic)
+
     userdata = {'args': args}
 
+    properties=Properties(PacketTypes.CONNECT)
+    properties.SessionExpiryInterval=3600
+    client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=client_id, protocol=mqtt.MQTTv5)
+    client.username_pw_set(username=args.client, password=password)
+    client.user_data_set(userdata)
+    #client.logger = logger
+
+    client.on_connect = on_connect
+    client.on_message = on_message
+    client.on_disconnect = on_disconnect
+
+    client.reconnect_delay_set(min_delay=1, max_delay=120)
+    client.connect(args.host, args.port, clean_start=False) #, properties=properties)
+
+    client.subscribe(topic, qos=args.qos)
+
     try:
-        subscribe.callback(receive_data, hostname=args.host, port=args.port,
-                           topics=f'{args.client.removesuffix("_ase")}/#',
-                           qos=args.qos, clean_session=False,
-                           auth=auth, client_id=f'{args.client.removesuffix("_ase")}_client_ase',
-                           userdata=userdata)
-    except (KeyboardInterrupt, Exception) as e:
-        logging.info(f"Terminating: ....{e}")
+        logger.info("Avvio del loop MQTT.")
+        client.loop_forever()
+    except KeyboardInterrupt:
+        logger.info("Terminazione manuale.")
+    except Exception as e:
+        logger.error(f"Errore durante il ciclo MQTT: {e}")
+    finally:
+        client.disconnect()
+        logger.info("Disconnesso dal broker.")
 
 if __name__ == "__main__":
     main()