53 lines
1.7 KiB
Python
53 lines
1.7 KiB
Python
import subprocess
|
|
import os
|
|
from config import settings
|
|
|
|
|
|
def register_device_password(serial_number: str, password: str) -> bool:
|
|
"""Register a device in the Mosquitto password file.
|
|
|
|
Uses mosquitto_passwd to add/update the device credentials.
|
|
The serial number is used as the MQTT username.
|
|
Returns True on success, False on failure.
|
|
"""
|
|
passwd_file = settings.mosquitto_password_file
|
|
|
|
# Ensure the password file exists
|
|
if not os.path.exists(passwd_file):
|
|
# Create the file if it doesn't exist
|
|
os.makedirs(os.path.dirname(passwd_file), exist_ok=True)
|
|
open(passwd_file, "a").close()
|
|
|
|
try:
|
|
# Use mosquitto_passwd with -b flag (batch mode) to set password
|
|
result = subprocess.run(
|
|
["mosquitto_passwd", "-b", passwd_file, serial_number, password],
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=10,
|
|
)
|
|
return result.returncode == 0
|
|
except (subprocess.TimeoutExpired, FileNotFoundError) as e:
|
|
print(f"[WARNING] Mosquitto password registration failed: {e}")
|
|
return False
|
|
|
|
|
|
def remove_device_password(serial_number: str) -> bool:
|
|
"""Remove a device from the Mosquitto password file."""
|
|
passwd_file = settings.mosquitto_password_file
|
|
|
|
if not os.path.exists(passwd_file):
|
|
return True
|
|
|
|
try:
|
|
result = subprocess.run(
|
|
["mosquitto_passwd", "-D", passwd_file, serial_number],
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=10,
|
|
)
|
|
return result.returncode == 0
|
|
except (subprocess.TimeoutExpired, FileNotFoundError) as e:
|
|
print(f"[WARNING] Mosquitto password removal failed: {e}")
|
|
return False
|