fix(python): Fixes for Python code scanning alerts

Author: lucasssvaz

.github/scripts/merge_packages.py

@@ -17,7 +17,8 @@
 
 
 def load_package(filename):
-    pkg = json.load(open(filename))["packages"][0]
+    with open(filename) as f:
+        pkg = json.load(f)["packages"][0]
     print("Loaded package {0} from {1}".format(pkg["name"], filename), file=sys.stderr)
     print("{0} platform(s), {1} tools".format(len(pkg["platforms"]), len(pkg["tools"])), file=sys.stderr)
     return pkg

libraries/ArduinoOTA/examples/BasicOTA/BasicOTA.ino

@@ -19,6 +19,7 @@
 
 const char *ssid = "..........";
 const char *password = "..........";
+uint32_t last_ota_time = 0;
 
 void setup() {
   Serial.begin(115200);
@@ -40,9 +41,13 @@ void setup() {
   // No authentication by default
   // ArduinoOTA.setPassword("admin");
 
-  // Password can be set with it's md5 value as well
-  // MD5(admin) = 21232f297a57a5a743894a0e4a801fc3
-  // ArduinoOTA.setPasswordHash("21232f297a57a5a743894a0e4a801fc3");
+  // Password can be set with plain text (will be hashed internally)
+  // The authentication uses PBKDF2-HMAC-SHA256 with 10,000 iterations
+  // ArduinoOTA.setPassword("admin");
+
+  // Or set password with pre-hashed value (SHA256 hash of "admin")
+  // SHA256(admin) = 8c6976e5b5410415bde908bd4dee15dfb167a9c873fc4bb8a81f6f2ab448a918
+  // ArduinoOTA.setPasswordHash("8c6976e5b5410415bde908bd4dee15dfb167a9c873fc4bb8a81f6f2ab448a918");
 
   ArduinoOTA
     .onStart([]() {
@@ -60,7 +65,10 @@ void setup() {
       Serial.println("\nEnd");
     })
     .onProgress([](unsigned int progress, unsigned int total) {
-      Serial.printf("Progress: %u%%\r", (progress / (total / 100)));
+      if (millis() - last_ota_time > 500) {
+        Serial.printf("Progress: %u%%\n", (progress / (total / 100)));
+        last_ota_time = millis();
+      }
     })
     .onError([](ota_error_t error) {
       Serial.printf("Error[%u]: ", error);

libraries/ArduinoOTA/src/ArduinoOTA.cpp

@@ -19,7 +19,8 @@
 #include "ArduinoOTA.h"
 #include "NetworkClient.h"
 #include "ESPmDNS.h"
-#include "MD5Builder.h"
+#include "SHA2Builder.h"
+#include "PBKDF2_HMACBuilder.h"
 #include "Update.h"
 
 // #define OTA_DEBUG Serial
@@ -72,18 +73,20 @@ String ArduinoOTAClass::getHostname() {
 
 ArduinoOTAClass &ArduinoOTAClass::setPassword(const char *password) {
   if (_state == OTA_IDLE && password) {
-    MD5Builder passmd5;
-    passmd5.begin();
-    passmd5.add(password);
-    passmd5.calculate();
+    // Hash the password with SHA256 for storage (not plain text)
+    SHA256Builder pass_hash;
+    pass_hash.begin();
+    pass_hash.add(password);
+    pass_hash.calculate();
     _password.clear();
-    _password = passmd5.toString();
+    _password = pass_hash.toString();
   }
   return *this;
 }
 
 ArduinoOTAClass &ArduinoOTAClass::setPasswordHash(const char *password) {
   if (_state == OTA_IDLE && password) {
+    // Store the pre-hashed password directly
     _password.clear();
     _password = password;
   }
@@ -188,17 +191,18 @@ void ArduinoOTAClass::_onRx() {
     _udp_ota.read();
     _md5 = readStringUntil('\n');
     _md5.trim();
-    if (_md5.length() != 32) {
+    if (_md5.length() != 32) {  // MD5 produces 32 character hex string for firmware integrity
       log_e("bad md5 length");
       return;
     }
 
     if (_password.length()) {
-      MD5Builder nonce_md5;
-      nonce_md5.begin();
-      nonce_md5.add(String(micros()));
-      nonce_md5.calculate();
-      _nonce = nonce_md5.toString();
+      // Generate a random challenge (nonce)
+      SHA256Builder nonce_sha256;
+      nonce_sha256.begin();
+      nonce_sha256.add(String(micros()) + String(random(1000000)));
+      nonce_sha256.calculate();
+      _nonce = nonce_sha256.toString();
 
       _udp_ota.beginPacket(_udp_ota.remoteIP(), _udp_ota.remotePort());
       _udp_ota.printf("AUTH %s", _nonce.c_str());
@@ -222,20 +226,37 @@ void ArduinoOTAClass::_onRx() {
     _udp_ota.read();
     String cnonce = readStringUntil(' ');
     String response = readStringUntil('\n');
-    if (cnonce.length() != 32 || response.length() != 32) {
+    if (cnonce.length() != 64 || response.length() != 64) {  // SHA256 produces 64 character hex string
       log_e("auth param fail");
       _state = OTA_IDLE;
       return;
     }
 
-    String challenge = _password + ":" + String(_nonce) + ":" + cnonce;
-    MD5Builder _challengemd5;
-    _challengemd5.begin();
-    _challengemd5.add(challenge);
-    _challengemd5.calculate();
-    String result = _challengemd5.toString();
-
-    if (result.equals(response)) {
+    // Verify the challenge/response using PBKDF2-HMAC-SHA256
+    // The client should derive a key using PBKDF2-HMAC-SHA256 with:
+    // - password: the OTA password (or its hash if using setPasswordHash)
+    // - salt: nonce + cnonce
+    // - iterations: 10000 (or configurable)
+    // Then hash the challenge with the derived key
+
+    String salt = _nonce + ":" + cnonce;
+    SHA256Builder sha256;
+    // Use the stored password hash for PBKDF2 derivation
+    PBKDF2_HMACBuilder pbkdf2(&sha256, _password, salt, 10000);
+
+    pbkdf2.begin();
+    pbkdf2.calculate();
+    String derived_key = pbkdf2.toString();
+
+    // Create challenge: derived_key + nonce + cnonce
+    String challenge = derived_key + ":" + _nonce + ":" + cnonce;
+    SHA256Builder challenge_sha256;
+    challenge_sha256.begin();
+    challenge_sha256.add(challenge);
+    challenge_sha256.calculate();
+    String expected_response = challenge_sha256.toString();
+
+    if (expected_response.equals(response)) {
       _udp_ota.beginPacket(_udp_ota.remoteIP(), _udp_ota.remotePort());
       _udp_ota.print("OK");
       _udp_ota.endPacket();
@@ -266,7 +287,8 @@ void ArduinoOTAClass::_runUpdate() {
     _state = OTA_IDLE;
     return;
   }
-  Update.setMD5(_md5.c_str());
+
+  Update.setMD5(_md5.c_str());  // Note: Update library still uses MD5 for firmware integrity, this is separate from authentication
 
   if (_start_callback) {
     _start_callback();

libraries/ArduinoOTA/src/ArduinoOTA.h

@@ -54,7 +54,7 @@ class ArduinoOTAClass {
   //Sets the password that will be required for OTA. Default NULL
   ArduinoOTAClass &setPassword(const char *password);
 
-  //Sets the password as above but in the form MD5(password). Default NULL
+  //Sets the password as above but in the form SHA256(password). Default NULL
   ArduinoOTAClass &setPasswordHash(const char *password);
 
   //Sets the partition label to write to when updating SPIFFS. Default NULL

libraries/WiFi/examples/WiFiUDPClient/udp_server.py

@@ -2,6 +2,93 @@
 # for messages from the ESP32 board and prints them
 import socket
 import sys
+import subprocess
+import platform
+
+def get_interface_ips():
+    """Get all available interface IP addresses"""
+    interface_ips = []
+
+    # Try using system commands to get interface IPs
+    system = platform.system().lower()
+
+    try:
+        if system == "darwin" or system == "linux":
+            # Use 'ifconfig' on macOS/Linux
+            result = subprocess.run(['ifconfig'], capture_output=True, text=True, timeout=5)
+            if result.returncode == 0:
+                lines = result.stdout.split('\n')
+                for line in lines:
+                    if 'inet ' in line and '127.0.0.1' not in line:
+                        # Extract IP address from ifconfig output
+                        parts = line.strip().split()
+                        for i, part in enumerate(parts):
+                            if part == 'inet':
+                                if i + 1 < len(parts):
+                                    ip = parts[i + 1]
+                                    if ip not in interface_ips and ip != '127.0.0.1':
+                                        interface_ips.append(ip)
+                                break
+        elif system == "windows":
+            # Use 'ipconfig' on Windows
+            result = subprocess.run(['ipconfig'], capture_output=True, text=True, timeout=5)
+            if result.returncode == 0:
+                lines = result.stdout.split('\n')
+                for line in lines:
+                    if 'IPv4 Address' in line and '127.0.0.1' not in line:
+                        # Extract IP address from ipconfig output
+                        if ':' in line:
+                            ip = line.split(':')[1].strip()
+                            if ip not in interface_ips and ip != '127.0.0.1':
+                                interface_ips.append(ip)
+    except (subprocess.TimeoutExpired, subprocess.SubprocessError, FileNotFoundError):
+        print("Error: Failed to get interface IPs using system commands")
+        print("Trying fallback methods...")
+
+    # Fallback: try to get IPs using socket methods
+    if not interface_ips:
+        try:
+            # Get all IP addresses associated with the hostname
+            hostname = socket.gethostname()
+            ip_list = socket.gethostbyname_ex(hostname)[2]
+            for ip in ip_list:
+                if ip not in interface_ips and ip != '127.0.0.1':
+                    interface_ips.append(ip)
+        except socket.gaierror:
+            print("Error: Failed to get interface IPs using sockets")
+
+    # Fail if no interfaces found
+    if not interface_ips:
+        print("Error: No network interfaces found. Please check your network configuration.")
+        sys.exit(1)
+
+    return interface_ips
+
+def select_interface(interface_ips):
+    """Ask user to select which interface to bind to"""
+    if len(interface_ips) == 1:
+        print(f"Using interface: {interface_ips[0]}")
+        return interface_ips[0]
+
+    print("Multiple network interfaces detected:")
+    for i, ip in enumerate(interface_ips, 1):
+        print(f"  {i}. {ip}")
+
+    while True:
+        try:
+            choice = input(f"Select interface (1-{len(interface_ips)}): ").strip()
+            choice_idx = int(choice) - 1
+            if 0 <= choice_idx < len(interface_ips):
+                selected_ip = interface_ips[choice_idx]
+                print(f"Selected interface: {selected_ip}")
+                return selected_ip
+            else:
+                print(f"Please enter a number between 1 and {len(interface_ips)}")
+        except ValueError:
+            print("Please enter a valid number")
+        except KeyboardInterrupt:
+            print("\nExiting...")
+            sys.exit(1)
 
 try:
     s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
@@ -10,15 +97,17 @@
     print("Failed to create socket. Error Code : " + str(msg[0]) + " Message " + msg[1])
     sys.exit()
 
+# Get available interfaces and let user choose
+interface_ips = get_interface_ips()
+selected_ip = select_interface(interface_ips)
+
 try:
-    s.bind(("", 3333))
+    s.bind((selected_ip, 3333))
 except socket.error as msg:
     print("Bind failed. Error: " + str(msg[0]) + ": " + msg[1])
     sys.exit()
 
-print("Server listening")
-
-print("Server listening")
+print(f"Server listening on {selected_ip}:3333")
 
 while 1:
     d = s.recvfrom(1024)

tools/espota.py

@@ -54,7 +54,6 @@
 # Constants
 PROGRESS_BAR_LENGTH = 60
 
-
 # update_progress(): Displays or updates a console progress bar
 def update_progress(progress):
     if PROGRESS:
@@ -94,7 +93,8 @@ def serve(remote_addr, local_addr, remote_port, local_port, password, filename,
         return 1
 
     content_size = os.path.getsize(filename)
-    file_md5 = hashlib.md5(open(filename, "rb").read()).hexdigest()
+    with open(filename, "rb") as f:
+        file_md5 = hashlib.md5(f.read()).hexdigest()
     logging.info("Upload size: %d", content_size)
     message = "%d %d %d %s\n" % (command, local_port, content_size, file_md5)
 
@@ -118,7 +118,7 @@ def serve(remote_addr, local_addr, remote_port, local_port, password, filename,
             return 1
         sock2.settimeout(TIMEOUT)
         try:
-            data = sock2.recv(37).decode()
+            data = sock2.recv(69).decode()  # "AUTH " + 64-char SHA256 nonce
             break
         except:  # noqa: E722
             sys.stderr.write(".")
@@ -132,18 +132,32 @@ def serve(remote_addr, local_addr, remote_port, local_port, password, filename,
     if data != "OK":
         if data.startswith("AUTH"):
             nonce = data.split()[1]
+
+            # Generate client nonce (cnonce)
             cnonce_text = "%s%u%s%s" % (filename, content_size, file_md5, remote_addr)
-            cnonce = hashlib.md5(cnonce_text.encode()).hexdigest()
-            passmd5 = hashlib.md5(password.encode()).hexdigest()
-            result_text = "%s:%s:%s" % (passmd5, nonce, cnonce)
-            result = hashlib.md5(result_text.encode()).hexdigest()
+            cnonce = hashlib.sha256(cnonce_text.encode()).hexdigest()
+
+            # PBKDF2-HMAC-SHA256 challenge/response protocol
+            # The ESP32 stores the password as SHA256 hash, so we need to hash the password first
+            # 1. Hash the password with SHA256 (to match ESP32 storage)
+            password_hash = hashlib.sha256(password.encode()).hexdigest()
+
+            # 2. Derive key using PBKDF2-HMAC-SHA256 with the password hash
+            salt = nonce + ":" + cnonce
+            derived_key = hashlib.pbkdf2_hmac('sha256', password_hash.encode(), salt.encode(), 10000)
+            derived_key_hex = derived_key.hex()
+
+            # 3. Create challenge response
+            challenge = derived_key_hex + ":" + nonce + ":" + cnonce
+            response = hashlib.sha256(challenge.encode()).hexdigest()
+
             sys.stderr.write("Authenticating...")
             sys.stderr.flush()
-            message = "%d %s %s\n" % (AUTH, cnonce, result)
+            message = "%d %s %s\n" % (AUTH, cnonce, response)
             sock2.sendto(message.encode(), remote_address)
             sock2.settimeout(10)
             try:
-                data = sock2.recv(32).decode()
+                data = sock2.recv(64).decode()  # SHA256 produces 64 character response
             except:  # noqa: E722
                 sys.stderr.write("FAIL\n")
                 logging.error("No Answer to our Authentication")
@@ -163,6 +177,7 @@ def serve(remote_addr, local_addr, remote_port, local_port, password, filename,
     sock2.close()
 
     logging.info("Waiting for device...")
+
     try:
         sock.settimeout(10)
         connection, client_address = sock.accept()
@@ -172,6 +187,7 @@ def serve(remote_addr, local_addr, remote_port, local_port, password, filename,
         logging.error("No response from device")
         sock.close()
         return 1
+
     try:
         with open(filename, "rb") as f:
             if PROGRESS:
@@ -225,7 +241,8 @@ def serve(remote_addr, local_addr, remote_port, local_port, password, filename,
             logging.error("Error response from device")
             connection.close()
             return 1
-
+    except Exception as e:  # noqa: E722
+        logging.error("Error: %s", str(e))
     finally:
         connection.close()