david
67bb68b097
Mosquitto 2.1.2 silently skips ACL file if world-readable or wrong owner. Listener now sets proper permissions on ACL file writes. Add CYD lock flash feedback on blocked tap.
51 lines
1.2 KiB
Python
51 lines
1.2 KiB
Python
#!/usr/bin/env -S uv run --script
|
|
# /// script
|
|
# dependencies = [
|
|
# "paho-mqtt>=2.0",
|
|
# ]
|
|
# ///
|
|
|
|
import os
|
|
import stat
|
|
|
|
import paho.mqtt.client as mqtt
|
|
|
|
MQTT_HOST = os.environ.get("MQTT_HOST", "mosquitto")
|
|
MQTT_PORT = int(os.environ.get("MQTT_PORT", "1883"))
|
|
ACL_PATH = "/mosquitto/config/acl.conf"
|
|
|
|
locked = False
|
|
|
|
def rebuild_acl():
|
|
lines = []
|
|
if locked:
|
|
lines.append("topic deny zigbee2mqtt/+/set")
|
|
lines.append("topic readwrite #")
|
|
with open(ACL_PATH, "w") as f:
|
|
f.write("\n".join(lines) + "\n")
|
|
os.chmod(ACL_PATH, stat.S_IRWXU)
|
|
os.chown(ACL_PATH, 1883, 1883)
|
|
os.system("docker kill -s HUP mosquitto 2>/dev/null || true")
|
|
|
|
def on_message(client, userdata, msg):
|
|
global locked
|
|
payload = msg.payload.decode().strip().upper()
|
|
if payload == "ON":
|
|
locked = True
|
|
elif payload == "OFF":
|
|
locked = False
|
|
else:
|
|
return
|
|
rebuild_acl()
|
|
|
|
def main():
|
|
rebuild_acl()
|
|
client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
|
|
client.on_message = on_message
|
|
client.connect(MQTT_HOST, MQTT_PORT, 60)
|
|
client.subscribe("party-lock/set", qos=1)
|
|
client.loop_forever()
|
|
|
|
if __name__ == "__main__":
|
|
main()
|