feat: check at launch of container,

+ other security fixes against xss or some exploits

Author: leonpwd

src/compare_json.py

@@ -6,8 +6,18 @@ def load_notes_json(filepath):
     if not os.path.exists(filepath):
         print(f"Le fichier {filepath} n'existe pas. Retourne une liste vide.")
         return []   
-    with open(filepath, "r", encoding="utf-8") as f:
-        return json.load(f)
+    try:
+        with open(filepath, "r", encoding="utf-8") as f:
+            data = json.load(f)
+            if not isinstance(data, list):
+                raise ValueError("Le fichier JSON doit contenir une liste")
+            return data
+    except json.JSONDecodeError as e:
+        print(f"Erreur lors de la lecture du JSON {filepath}: {e}")
+        return []
+    except Exception as e:
+        print(f"Erreur inattendue lors de la lecture de {filepath}: {e}")
+        return []
 
 def save_notes_json(data, filepath):
     data.replace('�', 'é').replace('é', 'é').replace('è', 'è').replace('�', 'Á')

src/env.py

@@ -25,6 +25,23 @@ def validate_ntfy_url(url):
 
 #* Ajout de la gestion du timezone
 TZ = os.getenv("TZ", "Europe/Paris")
+# Validation simple du fuseau horaire
+try:
+    from zoneinfo import ZoneInfo
+    ZoneInfo(TZ)  # Vérifie que le TZ est valide
+except Exception:
+    print(f"Erreur: Fuseau horaire '{TZ}' invalide. Utilisation de Europe/Paris par défaut.")
+    TZ = "Europe/Paris"
+
+#* Validation de CHECK_INTERVAL
+try:
+    CHECK_INTERVAL = int(os.getenv("CHECK_INTERVAL", "1800"))
+    if CHECK_INTERVAL < 30 or CHECK_INTERVAL > 86400:
+        print(f"Avertissement: CHECK_INTERVAL doit être entre 30s et 24h. Utilisation de 1800s.")
+        CHECK_INTERVAL = 1800
+except ValueError:
+    print("Erreur: CHECK_INTERVAL doit être un nombre. Utilisation de 1800s.")
+    CHECK_INTERVAL = 1800
 
 #! Chargement des variables d'environnement importantes
 #? URL des notes à surveiller

src/main.py

@@ -38,10 +38,11 @@ def get_notes_content():
         'Sec-Fetch-User': '?1',
         'Cache-Control': 'max-age=0'
     }
-    response = requests.get(URL, headers=headers, timeout=30)
+    response = requests.get(URL, headers=headers, timeout=30, verify=True)
 
     if response.status_code != 200:
-        raise Exception(f"Erreur lors de la récupération des notes: {response.status_code} - {response.text}")
+        error_text = response.text[:500]  # Limiter la longueur du message d'erreur
+        raise Exception(f"Erreur lors de la récupération des notes: {response.status_code} - {error_text}")
     response.raise_for_status()
     return response.text
 
@@ -139,26 +140,9 @@ def send_notification(change):
                 print()
                 print("DEBUG : {e}")
 
-def main():
-    while True:
-        now = datetime.datetime.now()
-        # Mode DEBUG : exécution toutes les 30 secondes, sans contrainte d'heure
-        if LOG_LEVEL == "DEBUG":
-            interval = 30
-        else:
-            # Si on est hors de la plage minuit-7h, on attend jusqu'à minuit
-            if not (0 <= now.hour < 3):
-                next_midnight = (now + datetime.timedelta(days=1)).replace(hour=0, minute=0, second=0, microsecond=0)
-                sleep_seconds = (next_midnight - now).total_seconds()
-                print(f"Hors plage horaire, dodo jusqu'à minuit ({next_midnight.strftime('%Y-%m-%d %H:%M:%S')})")
-                time.sleep(sleep_seconds)
-                continue
-            interval = CHECK_INTERVAL
-            # Si on est entre 1h20 et 1h40, on attend 2 minutes
-            now_tz = get_tz_time()
-            if now_tz.hour == 1 and 20 <= now_tz.minute < 40:
-                interval = 120
-        
+def check_notes():
+    """Fonction pour vérifier et traiter les notes"""
+    try:
         # Récupérer le contenu des notes
         content = get_notes_content()
         current_time = get_tz_time().strftime("%Y-%m-%d %H:%M:%S")
@@ -171,7 +155,7 @@ def main():
         parse.convert_notes_to_json(content, STORAGE_NOTES_JSON_2)
         if not os.path.exists(STORAGE_NOTES_JSON_2):
             print(f"Erreur : le fichier {STORAGE_NOTES_JSON_2} n'a pas été créé.")
-            continue
+            return
 
         new_notes = comparator.load_notes_json(STORAGE_NOTES_JSON_2)
 
@@ -194,6 +178,35 @@ def main():
             if os.path.exists(STORAGE_NOTES_JSON):
                 os.remove(STORAGE_NOTES_JSON)
             shutil.move(STORAGE_NOTES_JSON_2, STORAGE_NOTES_JSON)
+    except Exception as e:
+        print(f"Erreur lors de la vérification des notes: {e}")
+
+def main():
+    # Premier check automatique au lancement
+    print("🚀 Premier check automatique au lancement du conteneur...")
+    check_notes()
+    print()
+    
+    while True:
+        now = datetime.datetime.now()
+        # Mode DEBUG : exécution toutes les 30 secondes, sans contrainte d'heure
+        if LOG_LEVEL == "DEBUG":
+            interval = 30
+        else:
+            # Si on est hors de la plage minuit-7h, on attend jusqu'à minuit
+            if not (0 <= now.hour < 3):
+                next_midnight = (now + datetime.timedelta(days=1)).replace(hour=0, minute=0, second=0, microsecond=0)
+                sleep_seconds = (next_midnight - now).total_seconds()
+                print(f"Hors plage horaire, dodo jusqu'à minuit ({next_midnight.strftime('%Y-%m-%d %H:%M:%S')})")
+                time.sleep(sleep_seconds)
+                continue
+            interval = CHECK_INTERVAL
+            # Si on est entre 1h20 et 1h40, on attend 2 minutes
+            now_tz = get_tz_time()
+            if now_tz.hour == 1 and 20 <= now_tz.minute < 40:
+                interval = 120
+        
+        check_notes()
 
         next_time = get_tz_time() + datetime.timedelta(seconds=interval)
         print("Prochain check à", next_time.strftime("%Y-%m-%d %H:%M:%S"))

src/parse.py

@@ -46,24 +46,33 @@ def convert_notes_to_json(url_response, json_file):
         thead = soup.find("thead")
         if thead is None:
             print("Avertissement : balise <thead> non trouvée dans la réponse, le serveur est probablement en train de se reload, attente 1 minutes avant relance...")
-            # Sauvegarder en debug uniquement si LOG_LEVEL == DEBUG
             import os
             if os.getenv("LOG_LEVEL", "INFO").upper() == "DEBUG":
                 try:
                     with open("debug_last_notes.html", "w", encoding="utf-8") as f:
-                        f.write(html_content[:10000])  # Limiter à 10KB pour éviter les gros fichiers
-                    os.chmod("debug_last_notes.html", 0o600)  # Restreindre les permissions
+                        f.write(html_content[:10000])
+                    os.chmod("debug_last_notes.html", 0o600)
                 except Exception as e:
                     print(f"Impossible de sauvegarder le fichier de debug: {e}")
             time.sleep(60)
             print("Redémarrage du script...")
             sys.exit(1)
-        header_row = thead.find_all("tr")[1]
+        
+        # Vérifier que la structure est valide
+        tr_elements = thead.find_all("tr")
+        if len(tr_elements) < 2:
+            raise ValueError("Structure HTML invalide: <thead> ne contient pas assez de <tr>")
+        
+        header_row = tr_elements[1]
         headers = [fix_encoding_accents(th.get_text(separator=" ", strip=True).split("\n")[0]) for th in header_row.find_all("th")]
 
+        tbody = soup.find("tbody")
+        if tbody is None:
+            raise ValueError("Structure HTML invalide: <tbody> introuvable")
+        
         rows = [
             [fix_encoding_accents(td.get_text(separator=" ", strip=True)) for td in row.find_all("td")]
-            for row in soup.find("tbody").find_all("tr")
+            for row in tbody.find_all("tr")
             if "master-1" in row.get("class", [])
     ]