mirror of
https://github.com/Felitendo/moondeck-switchbot-wol.git
synced 2026-11-01 09:23:25 +01:00
423 lines
16 KiB
Python
423 lines
16 KiB
Python
"""Tests for the Python parts: the SwitchBot client, the MoonDeck settings
|
|
editor, both ends of the login trigger, and pairing. Nothing here touches the
|
|
real system: every path is in a temporary folder and the agent runs dry."""
|
|
|
|
import base64
|
|
import hashlib
|
|
import hmac
|
|
import http.server
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
import queue
|
|
import shutil
|
|
import socket
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import threading
|
|
import unittest
|
|
|
|
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|
LIB = os.path.join(ROOT, "src", "lib")
|
|
AGENT = os.path.join(ROOT, "src", "login-agent")
|
|
PAIR = os.path.join(LIB, "pair.py")
|
|
FIXTURE = os.path.join(ROOT, "tests", "moondeck-settings.json")
|
|
|
|
TOKEN = "a" * 96
|
|
SECRET = "b" * 32
|
|
TRIGGER = "0123456789abcdef0123456789abcdef"
|
|
|
|
|
|
class SwitchBotMock(http.server.BaseHTTPRequestHandler):
|
|
"""Answers like the real API, and checks the signature like it."""
|
|
|
|
commands = []
|
|
|
|
def signed(self):
|
|
expected = base64.b64encode(hmac.new(
|
|
SECRET.encode(),
|
|
(TOKEN + self.headers["t"] + self.headers["nonce"]).encode(),
|
|
hashlib.sha256).digest()).decode()
|
|
return self.headers["Authorization"] == TOKEN and self.headers["sign"] == expected
|
|
|
|
def answer(self, status, body):
|
|
data = json.dumps(body).encode()
|
|
self.send_response(status)
|
|
self.send_header("Content-Type", "application/json")
|
|
self.send_header("Content-Length", str(len(data)))
|
|
self.end_headers()
|
|
self.wfile.write(data)
|
|
|
|
def do_GET(self):
|
|
if not self.signed():
|
|
self.answer(401, {"message": "Unauthorized"})
|
|
else:
|
|
self.answer(200, {"statusCode": 100, "message": "success", "body": {"deviceList": [
|
|
{"deviceId": "C0FFEE000002", "deviceName": "Lamp", "deviceType": "Plug Mini (EU)"},
|
|
{"deviceId": "C0FFEE000001", "deviceName": "PC Power", "deviceType": "Bot"},
|
|
]}})
|
|
|
|
def do_POST(self):
|
|
length = int(self.headers.get("Content-Length", 0))
|
|
body = json.loads(self.rfile.read(length))
|
|
if not self.signed():
|
|
self.answer(401, {"message": "Unauthorized"})
|
|
elif "C0FFEE000001" not in self.path:
|
|
self.answer(200, {"statusCode": 152, "message": "device not found"})
|
|
else:
|
|
SwitchBotMock.commands.append(body["command"])
|
|
self.answer(200, {"statusCode": 100, "message": "success", "body": {}})
|
|
|
|
def log_message(self, *_):
|
|
pass
|
|
|
|
|
|
def start_mock():
|
|
server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), SwitchBotMock)
|
|
threading.Thread(target=server.serve_forever, daemon=True).start()
|
|
return server
|
|
|
|
|
|
class SwitchBotTest(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
cls.server = start_mock()
|
|
|
|
@classmethod
|
|
def tearDownClass(cls):
|
|
cls.server.shutdown()
|
|
|
|
def run_helper(self, *args, token=TOKEN):
|
|
env = dict(os.environ, MSW_SB_TOKEN=token, MSW_SB_SECRET=SECRET,
|
|
MSW_SB_API="http://127.0.0.1:%d/v1.1" % self.server.server_port)
|
|
return subprocess.run([sys.executable, os.path.join(LIB, "switchbot.py"), *args],
|
|
env=env, capture_output=True, text=True, check=False)
|
|
|
|
def test_devices_put_bots_first(self):
|
|
result = self.run_helper("devices")
|
|
self.assertEqual(result.returncode, 0)
|
|
self.assertEqual(result.stdout.splitlines()[0], "C0FFEE000001\tPC Power\tBot")
|
|
|
|
def test_command(self):
|
|
result = self.run_helper("command", "C0FFEE000001", "press")
|
|
self.assertEqual(result.returncode, 0)
|
|
self.assertEqual(SwitchBotMock.commands[-1], "press")
|
|
|
|
def test_wrong_token(self):
|
|
result = self.run_helper("devices", token="c" * 96)
|
|
self.assertEqual(result.returncode, 3)
|
|
self.assertIn("Unauthorized", result.stdout)
|
|
|
|
def test_api_error(self):
|
|
result = self.run_helper("command", "NOPE", "press")
|
|
self.assertEqual((result.returncode, result.stdout.strip()), (3, "152\tdevice not found"))
|
|
|
|
def test_unreachable(self):
|
|
env = dict(os.environ, MSW_SB_API="http://127.0.0.1:9/v1.1")
|
|
result = subprocess.run([sys.executable, os.path.join(LIB, "switchbot.py"), "devices"],
|
|
env=env, capture_output=True, text=True, check=False)
|
|
self.assertEqual(result.returncode, 2)
|
|
|
|
|
|
class MoonDeckTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.dir = tempfile.mkdtemp()
|
|
self.path = os.path.join(self.dir, "settings.json")
|
|
shutil.copy(FIXTURE, self.path)
|
|
with open(FIXTURE, encoding="utf-8") as handle:
|
|
self.original = json.load(handle)
|
|
self.host = self.original["currentHostId"]
|
|
|
|
def tearDown(self):
|
|
shutil.rmtree(self.dir)
|
|
|
|
def helper(self, *args):
|
|
result = subprocess.run([sys.executable, os.path.join(LIB, "moondeck.py"), self.path, *args],
|
|
capture_output=True, text=True, check=True)
|
|
return result.stdout
|
|
|
|
def load(self):
|
|
with open(self.path, encoding="utf-8") as handle:
|
|
return json.load(handle)["hostSettings"][self.host]
|
|
|
|
def test_hosts(self):
|
|
self.assertEqual(self.helper("hosts").rstrip("\n").split("\x1f"),
|
|
[self.host, "gaming-pc", "192.168.2.50", "1", "0", ""])
|
|
|
|
def test_point_and_restore(self):
|
|
change = self.helper("point", self.host, "/x/wake", "180")
|
|
host = self.load()
|
|
self.assertEqual(host["wolSettings"], {"useCustomWolExec": True, "sendOnce": True,
|
|
"customWolExecPath": "/x/wake", "port": 9})
|
|
self.assertEqual(host["runnerTimeouts"]["wakeOnLan"], 180)
|
|
|
|
self.helper("restore", change.strip())
|
|
with open(self.path, encoding="utf-8") as handle:
|
|
self.assertEqual(json.load(handle), self.original)
|
|
|
|
def test_restore_keeps_what_the_user_changed_since(self):
|
|
change = self.helper("point", self.host, "/x/wake", "180")
|
|
with open(self.path, encoding="utf-8") as handle:
|
|
data = json.load(handle)
|
|
data["hostSettings"][self.host]["runnerTimeouts"]["wakeOnLan"] = 300
|
|
with open(self.path, "w", encoding="utf-8") as handle:
|
|
json.dump(data, handle)
|
|
|
|
self.helper("restore", change.strip())
|
|
host = self.load()
|
|
self.assertEqual(host["runnerTimeouts"]["wakeOnLan"], 300)
|
|
self.assertFalse(host["wolSettings"]["useCustomWolExec"])
|
|
|
|
def test_a_longer_timeout_stays(self):
|
|
self.helper("point", self.host, "/x/wake", "30")
|
|
self.assertEqual(self.load()["runnerTimeouts"]["wakeOnLan"], 60)
|
|
|
|
|
|
class AgentTest(unittest.TestCase):
|
|
"""The agent on its own, then together with knock.py."""
|
|
|
|
def setUp(self):
|
|
self.dir = tempfile.mkdtemp()
|
|
self.conf = os.path.join(self.dir, "etc")
|
|
self.dm = os.path.join(self.dir, "plasmalogin.conf")
|
|
os.makedirs(self.conf)
|
|
with open(os.path.join(self.conf, "host.conf"), "w", encoding="utf-8") as handle:
|
|
handle.write("# test\nEnabled=yes\nUser=alice\nPort=58471\n")
|
|
with open(os.path.join(self.conf, "secret"), "w", encoding="utf-8") as handle:
|
|
handle.write(TRIGGER + "\n")
|
|
self.env = dict(os.environ, MSW_HOST_CONFDIR=self.conf,
|
|
MSW_HOST_STATEDIR=os.path.join(self.dir, "var"),
|
|
MSW_DM_CONFIG=self.dm, MSW_DRY_RUN="1")
|
|
|
|
def tearDown(self):
|
|
shutil.rmtree(self.dir)
|
|
|
|
def agent(self, *args, **kwargs):
|
|
return subprocess.run([sys.executable, AGENT, *args], env=self.env,
|
|
capture_output=True, text=True, check=False, **kwargs)
|
|
|
|
def write_dm(self, text):
|
|
with open(self.dm, "w", encoding="utf-8") as handle:
|
|
handle.write(text)
|
|
|
|
def read_dm(self):
|
|
with open(self.dm, encoding="utf-8") as handle:
|
|
return handle.read()
|
|
|
|
def arm(self):
|
|
# The agent's own functions, with the test's paths.
|
|
code = ("import importlib.machinery, importlib.util, sys; "
|
|
"loader = importlib.machinery.SourceFileLoader('agent', sys.argv[1]); "
|
|
"agent = importlib.util.module_from_spec(importlib.util.spec_from_loader('agent', loader)); "
|
|
"loader.exec_module(agent); "
|
|
"agent.arm(sys.argv[2], 'alice')")
|
|
subprocess.run([sys.executable, "-c", code, AGENT, self.dm], env=self.env, check=True,
|
|
capture_output=True)
|
|
|
|
def round_trip(self, before):
|
|
if before is not None:
|
|
self.write_dm(before)
|
|
self.arm()
|
|
armed = self.read_dm()
|
|
self.assertIn("User=alice", armed)
|
|
self.agent("--disarm")
|
|
if before is None:
|
|
self.assertFalse(os.path.exists(self.dm))
|
|
else:
|
|
self.assertEqual(self.read_dm(), before)
|
|
return armed
|
|
|
|
def test_section_with_session(self):
|
|
armed = self.round_trip("[Autologin]\nSession=plasma\n")
|
|
self.assertEqual(armed, "[Autologin]\nUser=alice\nSession=plasma\n")
|
|
|
|
def test_user_already_set(self):
|
|
self.round_trip("[General]\nNumlock=on\n\n[Autologin]\nUser=bob\n")
|
|
|
|
def test_no_section(self):
|
|
self.round_trip("[General]\nNumlock=on\n")
|
|
|
|
def test_no_file(self):
|
|
self.round_trip(None)
|
|
|
|
def test_check(self):
|
|
result = self.agent("--check")
|
|
self.assertEqual((result.returncode, result.stdout.strip()), (0, self.dm))
|
|
|
|
def serve(self):
|
|
"""Starts the agent for every connection, the way the socket unit does."""
|
|
listener = socket.socket()
|
|
listener.bind(("127.0.0.1", 0))
|
|
listener.listen()
|
|
|
|
def accept():
|
|
while True:
|
|
try:
|
|
connection, _ = listener.accept()
|
|
except OSError:
|
|
return
|
|
with connection:
|
|
subprocess.run([sys.executable, AGENT], env=self.env, stdin=connection,
|
|
stdout=connection, stderr=subprocess.DEVNULL, check=False)
|
|
|
|
threading.Thread(target=accept, daemon=True).start()
|
|
self.addCleanup(listener.close)
|
|
return listener.getsockname()[1]
|
|
|
|
def knock(self, port, secret):
|
|
env = dict(os.environ, MSW_TRIGGER_SECRET=secret)
|
|
# The first address answers nothing, the second is the agent.
|
|
return subprocess.run([sys.executable, os.path.join(LIB, "knock.py"), str(port), "20",
|
|
"127.0.0.2", "127.0.0.1"], env=env, capture_output=True,
|
|
text=True, check=False, timeout=60)
|
|
|
|
def test_knock_logs_in_once(self):
|
|
self.write_dm("[Autologin]\nSession=plasma\n")
|
|
port = self.serve()
|
|
result = self.knock(port, TRIGGER)
|
|
self.assertEqual(result.returncode, 0, result.stdout)
|
|
self.assertIn("127.0.0.1 logs in now", result.stdout)
|
|
# Armed, logged in (dry run) and taken back out again.
|
|
self.assertEqual(self.read_dm(), "[Autologin]\nSession=plasma\n")
|
|
with open(os.path.join(self.dir, "var", "last"), encoding="utf-8") as handle:
|
|
self.assertTrue(handle.read().endswith("\tlogin\n"))
|
|
|
|
def test_knock_with_the_wrong_secret(self):
|
|
self.write_dm("[Autologin]\nSession=plasma\n")
|
|
port = self.serve()
|
|
result = self.knock(port, "f" * 32)
|
|
self.assertEqual(result.returncode, 1)
|
|
self.assertIn("does not know this Deck", result.stdout)
|
|
self.assertEqual(self.read_dm(), "[Autologin]\nSession=plasma\n")
|
|
|
|
|
|
def load_pair():
|
|
spec = importlib.util.spec_from_file_location("pair", PAIR)
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
def free_port():
|
|
with socket.socket() as probe:
|
|
probe.bind(("127.0.0.1", 0))
|
|
return probe.getsockname()[1]
|
|
|
|
|
|
class Pairing:
|
|
"""pair.py in the background, line by line."""
|
|
|
|
def __init__(self, test, *args, code=""):
|
|
self.process = subprocess.Popen(
|
|
[sys.executable, PAIR, *map(str, args)], env=dict(os.environ, MSW_PAIR_CODE=code),
|
|
stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, text=True)
|
|
self.lines = queue.Queue()
|
|
threading.Thread(target=self.read, daemon=True).start()
|
|
test.addCleanup(self.stop)
|
|
|
|
def read(self):
|
|
for line in self.process.stdout:
|
|
self.lines.put(line.split())
|
|
self.lines.put(["END"])
|
|
|
|
def next(self):
|
|
return self.lines.get(timeout=30)
|
|
|
|
def say(self, answer):
|
|
self.process.stdin.write(answer + "\n")
|
|
self.process.stdin.flush()
|
|
|
|
def stop(self):
|
|
self.process.kill()
|
|
self.process.wait()
|
|
self.process.stdin.close()
|
|
self.process.stdout.close()
|
|
|
|
|
|
class PairTest(unittest.TestCase):
|
|
CODE = "msw1:" + base64.b64encode(("TriggerSecret=%s\n" % TRIGGER).encode()).decode()
|
|
|
|
def deck(self):
|
|
"""A Deck that waits, once a scan finds it."""
|
|
port = free_port()
|
|
deck = Pairing(self, "listen", port, 30)
|
|
scan = subprocess.run([sys.executable, PAIR, "scan", str(port), "20", "127.0.0.1"],
|
|
capture_output=True, text=True, check=False, timeout=60)
|
|
self.assertEqual(scan.stdout.split()[:2], ["FOUND", "127.0.0.1"])
|
|
return deck, port
|
|
|
|
def meet(self):
|
|
"""A Deck and a PC that both show the same digits."""
|
|
deck, port = self.deck()
|
|
pc = Pairing(self, "send", "127.0.0.1", port, code=self.CODE)
|
|
pc_peer, deck_peer = pc.next(), deck.next()
|
|
self.assertEqual(pc_peer[0], "PEER")
|
|
self.assertEqual(deck_peer[:3], ["PEER", pc_peer[1], "127.0.0.1"])
|
|
return deck, pc
|
|
|
|
def test_pairing(self):
|
|
deck, pc = self.meet()
|
|
deck.say("yes")
|
|
pc.say("yes")
|
|
self.assertEqual(deck.next(), ["DATA", self.CODE])
|
|
deck.say("ok")
|
|
self.assertEqual(deck.next(), ["DONE"])
|
|
self.assertEqual(pc.next(), ["OK"])
|
|
|
|
def test_the_deck_says_no(self):
|
|
deck, pc = self.meet()
|
|
deck.say("no")
|
|
pc.say("yes")
|
|
self.assertEqual(pc.next(), ["NO"])
|
|
|
|
def test_the_pc_says_no(self):
|
|
deck, pc = self.meet()
|
|
deck.say("yes")
|
|
pc.say("no")
|
|
self.assertEqual(deck.next(), ["CANCELLED"])
|
|
|
|
def test_a_key_other_than_the_one_committed_to(self):
|
|
pair = load_pair()
|
|
deck, port = self.deck()
|
|
_, committed = pair.keypair()
|
|
_, sent = pair.keypair()
|
|
with socket.create_connection(("127.0.0.1", port), timeout=10) as connection:
|
|
stream = connection.makefile("rw", encoding="ascii", newline="\n")
|
|
stream.readline()
|
|
pair.send(stream, "COMMIT %s mallory" % pair.commitment(committed))
|
|
stream.readline()
|
|
pair.send(stream, "KEY %x" % sent)
|
|
self.assertEqual(stream.readline(), "")
|
|
# Turned away without asking, and still waiting for the real PC.
|
|
pc = Pairing(self, "send", "127.0.0.1", port, code=self.CODE)
|
|
self.assertEqual(deck.next()[0], "PEER")
|
|
self.assertEqual(pc.next()[0], "PEER")
|
|
|
|
def test_changed_on_the_way(self):
|
|
pair = load_pair()
|
|
encrypt, sign, _ = pair.derive(5, 7, 11, 13)
|
|
sealed = pair.seal(encrypt, sign, self.CODE)
|
|
self.assertEqual(pair.unseal(encrypt, sign, sealed), self.CODE)
|
|
data, tag = sealed.split(" ")
|
|
changed = base64.b64encode(bytes([base64.b64decode(data)[0] ^ 1]) + base64.b64decode(data)[1:])
|
|
self.assertIsNone(pair.unseal(encrypt, sign, "%s %s" % (changed.decode(), tag)))
|
|
|
|
def test_the_prime(self):
|
|
pair = load_pair()
|
|
# A typo in it would not survive these.
|
|
self.assertEqual(pow(2, pair.P - 1, pair.P), 1)
|
|
self.assertEqual(pow(2, (pair.P - 1) // 2 - 1, (pair.P - 1) // 2), 1)
|
|
|
|
def test_targets(self):
|
|
pair = load_pair()
|
|
home = pair.targets(["192.168.2.10/24"])
|
|
self.assertEqual((len(home), home[0], home[-1]), (254, "192.168.2.1", "192.168.2.254"))
|
|
big = pair.targets(["10.1.2.3/8"])
|
|
self.assertEqual((len(big), big[0]), (254, "10.1.2.1"))
|
|
self.assertEqual(pair.targets(["127.0.0.1"]), ["127.0.0.1"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|