2025-07-08 17:30:46 +02:00
|
|
|
#!/usr/bin/env python3
|
|
|
|
|
|
|
|
|
|
from mosq_test_helper import *
|
|
|
|
|
import http.client
|
|
|
|
|
import json
|
|
|
|
|
|
|
|
|
|
def write_config(filename, mqtt_port, http_port):
|
|
|
|
|
with open(filename, 'w') as f:
|
2025-10-14 01:18:37 +02:00
|
|
|
f.write("allow_anonymous true\n")
|
2025-07-08 17:30:46 +02:00
|
|
|
f.write(f"listener {mqtt_port}\n")
|
|
|
|
|
f.write(f"listener {http_port} 127.0.0.1\n")
|
|
|
|
|
f.write("protocol http_api\n")
|
|
|
|
|
f.write(f"http_dir ./\n")
|
|
|
|
|
|
|
|
|
|
def write_index():
|
|
|
|
|
with open("index.html", 'w') as f:
|
|
|
|
|
f.write("<html></html>")
|
|
|
|
|
|
|
|
|
|
mqtt_port, http_port = mosq_test.get_port(2)
|
|
|
|
|
conf_file = os.path.basename(__file__).replace('.py', '.conf')
|
|
|
|
|
write_config(conf_file, mqtt_port, http_port)
|
|
|
|
|
write_index()
|
|
|
|
|
|
|
|
|
|
broker = mosq_test.start_broker(filename=os.path.basename(__file__), use_conf=True, port=mqtt_port)
|
|
|
|
|
|
|
|
|
|
rc = 1
|
|
|
|
|
|
|
|
|
|
try:
|
|
|
|
|
http_conn = http.client.HTTPConnection(f"localhost:{http_port}")
|
|
|
|
|
|
|
|
|
|
# Bad request
|
|
|
|
|
http_conn.request("POST", "/post")
|
|
|
|
|
response = http_conn.getresponse()
|
|
|
|
|
if response.status != 405:
|
2025-10-14 01:18:37 +02:00
|
|
|
raise ValueError(f"Error: /post {response.status}")
|
2025-07-08 17:30:46 +02:00
|
|
|
|
|
|
|
|
# Bad request
|
|
|
|
|
http_conn.request("PUT", "/put")
|
|
|
|
|
response = http_conn.getresponse()
|
|
|
|
|
if response.status != 405:
|
2025-10-14 01:18:37 +02:00
|
|
|
raise ValueError(f"Error: /put {response.status}")
|
2025-07-08 17:30:46 +02:00
|
|
|
|
|
|
|
|
# Missing file
|
|
|
|
|
http_conn.request("GET", "/missing")
|
|
|
|
|
response = http_conn.getresponse()
|
|
|
|
|
if response.status != 404:
|
2025-10-14 01:18:37 +02:00
|
|
|
raise ValueError(f"Error: /api/missing {response.status}")
|
2025-07-08 17:30:46 +02:00
|
|
|
|
|
|
|
|
# File not in dir
|
|
|
|
|
http_conn.request("GET", "../../../../../../../../etc/passwd")
|
|
|
|
|
response = http_conn.getresponse()
|
|
|
|
|
if response.status != 404:
|
2025-10-14 01:18:37 +02:00
|
|
|
raise ValueError(f"Error: ../../../../../../../../etc/passwd {response.status}")
|
2025-07-08 17:30:46 +02:00
|
|
|
|
|
|
|
|
# Present file
|
|
|
|
|
http_conn.request("GET", "/index.html")
|
|
|
|
|
response = http_conn.getresponse()
|
|
|
|
|
if response.status != 200:
|
2025-10-14 01:18:37 +02:00
|
|
|
raise ValueError(f"Error: /index.html {response.status}")
|
2025-07-08 17:30:46 +02:00
|
|
|
|
|
|
|
|
# Root
|
|
|
|
|
http_conn.request("GET", "/")
|
|
|
|
|
response = http_conn.getresponse()
|
|
|
|
|
if response.status != 200:
|
2025-10-14 01:18:37 +02:00
|
|
|
raise ValueError(f"Error: / {response.status}")
|
2025-07-08 17:30:46 +02:00
|
|
|
payload = response.read().decode('utf-8')
|
|
|
|
|
if payload != "<html></html>":
|
2025-10-14 01:18:37 +02:00
|
|
|
raise ValueError(f"Error: / {payload}")
|
2025-07-08 17:30:46 +02:00
|
|
|
|
|
|
|
|
|
|
|
|
|
rc = 0
|
|
|
|
|
except mosq_test.TestError:
|
|
|
|
|
pass
|
|
|
|
|
except Exception as e:
|
|
|
|
|
print(e)
|
|
|
|
|
finally:
|
|
|
|
|
os.remove(conf_file)
|
|
|
|
|
os.remove("index.html")
|
|
|
|
|
broker.terminate()
|
|
|
|
|
if mosq_test.wait_for_subprocess(broker):
|
|
|
|
|
print("broker not terminated")
|
|
|
|
|
if rc == 0: rc=1
|
|
|
|
|
(stdo, stde) = broker.communicate()
|
|
|
|
|
if rc != 0:
|
|
|
|
|
print(stde.decode('utf-8'))
|
|
|
|
|
rc = 1
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
exit(rc)
|
