40 lines
2.8 KiB
Python
40 lines
2.8 KiB
Python
#!/usr/bin/env python3
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
import urllib.request
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).parent
|
|
with tempfile.TemporaryDirectory() as temp:
|
|
env = {**os.environ, "LIGHTHOUSE_DB": f"{temp}/lighthouse.db", "LIGHTHOUSE_ADMIN_TOKEN": "test-token", "LIGHTHOUSE_PORT": "48031"}
|
|
process = subprocess.Popen([sys.executable, "lighthouse.py"], cwd=ROOT, env=env, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
|
|
try:
|
|
for _ in range(30):
|
|
try:
|
|
assert json.load(urllib.request.urlopen("http://127.0.0.1:48031/health", timeout=1))["execution"] == "disabled"
|
|
break
|
|
except OSError:
|
|
time.sleep(.1)
|
|
else: raise AssertionError("server did not start")
|
|
request = urllib.request.Request("http://127.0.0.1:48031/v1/intakes", data=json.dumps({"human_name":"Test","email":"test@example.invalid","server_ip":"203.0.113.8","domain_id":"DOMAIN-ZS"}).encode(), method="POST", headers={"X-Lighthouse-Admin-Token":"test-token"})
|
|
assert json.load(urllib.request.urlopen(request))["state"] == "PENDING_REVIEW"
|
|
bootstrap = urllib.request.Request("http://127.0.0.1:48031/v1/nodes/bootstrap", data=json.dumps({"id":"NODE-TEST-001","domain_id":"DOMAIN-ZS","display_name":"Test entry","server_ip":"203.0.113.8"}).encode(), method="POST", headers={"X-Lighthouse-Admin-Token":"test-token"})
|
|
assert json.load(urllib.request.urlopen(bootstrap))["state"] == "CONNECTED_PENDING_CONNECTOR"
|
|
preflight = urllib.request.Request("http://127.0.0.1:48031/v1/preflight", data=json.dumps({"action":"health_check","target_node_id":"NODE-TEST-001"}).encode(), method="POST", headers={"X-Lighthouse-Admin-Token":"test-token"})
|
|
assert json.load(urllib.request.urlopen(preflight))["decision"] == "REJECT"
|
|
fifth = urllib.request.Request("http://127.0.0.1:48031/v1/nodes/bootstrap", data=json.dumps({"id":"NODE-FIFTH-001","domain_id":"DOMAIN-FIFTH","display_name":"Not allowed","server_ip":"203.0.113.8"}).encode(), method="POST", headers={"X-Lighthouse-Admin-Token":"test-token"})
|
|
try: urllib.request.urlopen(fifth)
|
|
except urllib.error.HTTPError as error: assert error.code == 403
|
|
else: raise AssertionError("private fifth domain was accepted by enterprise lighthouse")
|
|
bad = urllib.request.Request("http://127.0.0.1:48031/v1/intakes", data=b'{"cmd":"rm -rf /"}', method="POST", headers={"X-Lighthouse-Admin-Token":"test-token"})
|
|
try: urllib.request.urlopen(bad)
|
|
except urllib.error.HTTPError as error: assert error.code == 400
|
|
else: raise AssertionError("raw command was not rejected")
|
|
finally:
|
|
process.terminate(); process.wait(timeout=5)
|
|
print("enterprise lighthouse tests passed")
|