You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
210 lines
8.0 KiB
210 lines
8.0 KiB
import io
|
|
import json
|
|
import unittest
|
|
from unittest.mock import Mock
|
|
|
|
|
|
class FakeRequest:
|
|
def __init__(self, path, payload=None):
|
|
self.postpath = [part.encode("utf-8") for part in path.strip("/").split("/") if part]
|
|
self.content = io.BytesIO(
|
|
b"" if payload is None else json.dumps(payload).encode("utf-8")
|
|
)
|
|
self.code = None
|
|
self.headers = {}
|
|
|
|
def setResponseCode(self, code):
|
|
self.code = code
|
|
|
|
def setHeader(self, name, value):
|
|
self.headers[name] = value
|
|
|
|
def getHeader(self, name):
|
|
if name == "content-length":
|
|
return str(len(self.content.getvalue()))
|
|
return None
|
|
|
|
|
|
class APITest(unittest.TestCase):
|
|
def setUp(self):
|
|
try:
|
|
import twisted.web.resource # noqa: F401
|
|
except ModuleNotFoundError as exc:
|
|
self.skipTest(f"Twisted is not installed: {exc}")
|
|
import API
|
|
|
|
self.api = API
|
|
self.peer_id = (1234567).to_bytes(4, "big")
|
|
self.config = {
|
|
"GLOBAL": {"SYSTEM_API_KEY": "system-secret", "_KILL_SERVER": False},
|
|
"SYSTEMS": {
|
|
"MASTER-A": {
|
|
"MODE": "MASTER",
|
|
"PEERS": {self.peer_id: {}},
|
|
"_opt_key": "peer-secret",
|
|
},
|
|
"OBP-A": {
|
|
"MODE": "OPENBRIDGE",
|
|
"PEERS": {},
|
|
},
|
|
},
|
|
}
|
|
self.bridges = {}
|
|
self.controller = API.FD_APIController(self.config, self.bridges)
|
|
|
|
def test_getoptions_returns_clear_no_options_response(self):
|
|
result = self.controller.getoptions("MASTER-A")
|
|
|
|
self.assertEqual(
|
|
result,
|
|
{"connected": True, "has_options": False, "options": ""},
|
|
)
|
|
|
|
def test_getoptions_decodes_byte_options_for_json(self):
|
|
self.config["SYSTEMS"]["MASTER-A"]["OPTIONS"] = b"KEY=peer-secret;TS1=91"
|
|
|
|
result = self.controller.getoptions("MASTER-A")
|
|
|
|
self.assertEqual(result["options"], "KEY=peer-secret;TS1=91")
|
|
self.assertTrue(result["has_options"])
|
|
|
|
def test_setoptions_stores_full_options_string_unchanged(self):
|
|
options = "KEY=peer-secret;TS1=91;DIAL=2350"
|
|
|
|
self.controller.options("MASTER-A", options)
|
|
|
|
self.assertEqual(self.config["SYSTEMS"]["MASTER-A"]["OPTIONS"], options)
|
|
|
|
def test_user_reset_is_allowed_only_for_matching_peer_key(self):
|
|
system = self.controller.validateKey(1234567, "peer-secret")
|
|
|
|
self.assertEqual(system, "MASTER-A")
|
|
self.controller.reset(system)
|
|
self.assertTrue(self.config["SYSTEMS"]["MASTER-A"]["_reset"])
|
|
self.assertFalse(self.controller.validateKey(1234567, "wrong"))
|
|
|
|
def test_system_kill_sets_existing_control_flag(self):
|
|
self.assertTrue(self.controller.validateSystemKey("system-secret"))
|
|
|
|
self.controller.killserver()
|
|
|
|
self.assertTrue(self.config["GLOBAL"]["_KILL_SERVER"])
|
|
|
|
def test_options_get_endpoint_returns_json(self):
|
|
resource = self.api.make_api_resource(self.config, self.bridges)
|
|
request = FakeRequest(
|
|
"/api/v1/options/get",
|
|
{"dmrid": 1234567, "key": "peer-secret"},
|
|
)
|
|
|
|
body = resource.render_POST(request)
|
|
|
|
self.assertEqual(request.code, 200)
|
|
self.assertEqual(
|
|
json.loads(body.decode("utf-8")),
|
|
{"ok": True, "connected": True, "has_options": False, "options": ""},
|
|
)
|
|
|
|
def test_options_get_endpoint_rejects_bad_key(self):
|
|
resource = self.api.make_api_resource(self.config, self.bridges)
|
|
request = FakeRequest(
|
|
"/api/v1/options/get",
|
|
{"dmrid": 1234567, "key": "wrong"},
|
|
)
|
|
|
|
body = resource.render_POST(request)
|
|
|
|
self.assertEqual(request.code, 401)
|
|
self.assertEqual(
|
|
json.loads(body.decode("utf-8")),
|
|
{"ok": False, "error": "invalid_credentials"},
|
|
)
|
|
|
|
def test_endpoint_rejects_large_request_body(self):
|
|
resource = self.api.make_api_resource(self.config, self.bridges)
|
|
request = FakeRequest(
|
|
"/api/v1/options/set",
|
|
{"dmrid": 1234567, "key": "peer-secret", "options": "A" * 9000},
|
|
)
|
|
|
|
body = resource.render_POST(request)
|
|
|
|
self.assertEqual(request.code, 413)
|
|
self.assertEqual(
|
|
json.loads(body.decode("utf-8")),
|
|
{"ok": False, "error": "request_too_large"},
|
|
)
|
|
|
|
def test_dial_dispatches_only_to_connected_peer_with_system_key(self):
|
|
self.config["SYSTEMS"]["MASTER-A"]["PEERS"][self.peer_id]["CONNECTION"] = "YES"
|
|
control = Mock(return_value={"tg": 235})
|
|
resource = self.api.make_api_resource(self.config, self.bridges, control)
|
|
for action in ("link", "unlink", "status", "reset", "reset_tgs"):
|
|
payload = {"systemkey": "system-secret", "dmrid": 1234567,
|
|
"slot": 1, "action": action}
|
|
if action == "link":
|
|
payload["tg"] = 235
|
|
request = FakeRequest("/api/v1/dial", payload)
|
|
result = json.loads(resource.render_POST(request))
|
|
self.assertEqual(request.code, 200)
|
|
self.assertTrue(result["ok"])
|
|
control.assert_called_with("MASTER-A", 1, action, payload.get("tg"))
|
|
|
|
def test_dial_rejects_invalid_input_without_dispatch(self):
|
|
control = Mock()
|
|
resource = self.api.make_api_resource(self.config, self.bridges, control)
|
|
base = {"systemkey": "system-secret", "dmrid": 1234567,
|
|
"slot": 2, "action": "link", "tg": 235}
|
|
cases = (
|
|
({"systemkey": None}, 401), ({"systemkey": ""}, 401),
|
|
({"systemkey": "peer-secret"}, 401), ({"slot": True}, 400),
|
|
({"slot": 3}, 400), ({"dmrid": -1}, 400),
|
|
({"dmrid": True}, 400), ({"dmrid": 2**32}, 400),
|
|
({"action": "kill"}, 400), ({"action": []}, 400),
|
|
({"tg": True}, 400), ({"tg": "235"}, 400),
|
|
({"tg": 1000000}, 400), ({"action": "reset"}, 400),
|
|
({"action": "reset_tgs"}, 400),
|
|
({}, 404),
|
|
)
|
|
for changes, code in cases:
|
|
with self.subTest(changes=changes):
|
|
request = FakeRequest("/api/v1/dial", {**base, **changes})
|
|
result = json.loads(resource.render_POST(request))
|
|
self.assertEqual(request.code, code)
|
|
self.assertFalse(result["ok"])
|
|
control.assert_not_called()
|
|
|
|
def test_dial_rejects_missing_configured_key_and_ambiguous_peer(self):
|
|
control = Mock()
|
|
resource = self.api.make_api_resource(self.config, self.bridges, control)
|
|
payload = {"dmrid": 1234567, "slot": 2, "action": "status"}
|
|
self.config["GLOBAL"].pop("SYSTEM_API_KEY")
|
|
request = FakeRequest("/api/v1/dial", payload)
|
|
resource.render_POST(request)
|
|
self.assertEqual(request.code, 401)
|
|
self.config["GLOBAL"]["SYSTEM_API_KEY"] = "system-secret"
|
|
self.config["SYSTEMS"]["MASTER-A"]["PEERS"][self.peer_id]["CONNECTION"] = "YES"
|
|
self.config["SYSTEMS"]["MASTER-B"] = self.config["SYSTEMS"]["MASTER-A"]
|
|
request = FakeRequest("/api/v1/dial", {**payload, "systemkey": "system-secret"})
|
|
resource.render_POST(request)
|
|
self.assertEqual(request.code, 409)
|
|
control.assert_not_called()
|
|
|
|
def test_dial_does_not_control_other_peers_sharing_a_master(self):
|
|
self.config['SYSTEMS']['MASTER-A']['PEERS'] = {
|
|
self.peer_id: {'CONNECTION': 'YES'},
|
|
(7654321).to_bytes(4, 'big'): {'CONNECTION': 'YES'},
|
|
}
|
|
control = Mock()
|
|
resource = self.api.make_api_resource(self.config, self.bridges, control)
|
|
request = FakeRequest('/api/v1/dial', {
|
|
'systemkey': 'system-secret', 'dmrid': 1234567, 'slot': 1, 'action': 'unlink',
|
|
})
|
|
result = json.loads(resource.render_POST(request))
|
|
self.assertEqual((request.code, result['error']), (409, 'shared_master'))
|
|
control.assert_not_called()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|