coreci
7f7868f7cf
---ci--- phase: 5 milestone: v0.1 status: complete requirements: covered: [REQ-01, REQ-02, REQ-03, REQ-04, REQ-05, REQ-06, REQ-07, REQ-08, REQ-09, REQ-10, REQ-11, REQ-12, REQ-13, REQ-14, REQ-15, REQ-16, REQ-17, REQ-18, REQ-19, REQ-20] partial: [] ---/ci---
207 lines
8.8 KiB
Python
207 lines
8.8 KiB
Python
import base64
|
|
import contextlib
|
|
import errno
|
|
import io
|
|
import json
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
import urllib.error
|
|
import urllib.request
|
|
import uuid as uuidlib
|
|
from unittest import mock
|
|
|
|
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), os.pardir)))
|
|
|
|
import coreci_broker
|
|
|
|
ADMIN_USER = "admin"
|
|
ADMIN_PASS = "correct-horse-battery-staple"
|
|
REGISTER_PATH = "/api/broker/agents/register"
|
|
APPS_PATH = "/api/broker/app_registrations"
|
|
USERS_PATH = "/api/broker/user_allowlist"
|
|
AGENT_ID_BODY = {"error": 400, "message": "invalid agent_id"}
|
|
TAGS_BODY = {"error": 400, "message": "invalid tags"}
|
|
USER_ID_BODY = {"error": 400, "message": "invalid user_id"}
|
|
|
|
|
|
def _basic_header(username, password):
|
|
raw = "%s:%s" % (username, password)
|
|
return "Basic " + base64.b64encode(raw.encode("utf-8")).decode("ascii")
|
|
|
|
|
|
class P5RegisterValidationTests(unittest.TestCase):
|
|
def setUp(self):
|
|
tmp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(tmp.cleanup)
|
|
self.tmp = tmp.name
|
|
self.env = {
|
|
"BROKER_ENV_PATH": os.path.join(self.tmp, ".env"),
|
|
"BROKER_DB_PATH": os.path.join(self.tmp, "broker.db"),
|
|
"BROKER_AUDIT_LOG_PATH": os.path.join(self.tmp, "broker.audit.log"),
|
|
"BROKER_ADMIN_USER": ADMIN_USER,
|
|
"BROKER_ADMIN_PASS": ADMIN_PASS,
|
|
}
|
|
with contextlib.redirect_stdout(io.StringIO()):
|
|
self.broker = coreci_broker.boot(dict(self.env))
|
|
self.addCleanup(self.broker.audit.close)
|
|
self.addCleanup(self.broker.store.close)
|
|
self.server = coreci_broker.serve("127.0.0.1", 0, self.broker)
|
|
self.addCleanup(self.server.shutdown)
|
|
self.addCleanup(self.server.server_close)
|
|
self.base = "http://127.0.0.1:%d" % self.server.server_address[1]
|
|
self.admin = {"Authorization": _basic_header(ADMIN_USER, ADMIN_PASS)}
|
|
self.app_uuid = str(uuidlib.uuid4())
|
|
self.user_id = "ceo"
|
|
self._seed()
|
|
|
|
def _request(self, method, path, payload=None, headers=None):
|
|
data = None if payload is None else json.dumps(payload).encode("utf-8")
|
|
request = urllib.request.Request(self.base + path, data=data, method=method)
|
|
for key, value in (headers or {}).items():
|
|
request.add_header(key, value)
|
|
try:
|
|
with urllib.request.urlopen(request, timeout=10) as response:
|
|
return response.status, json.loads(response.read().decode("utf-8"))
|
|
except urllib.error.HTTPError as error:
|
|
return error.code, json.loads(error.read().decode("utf-8"))
|
|
|
|
def _seed(self):
|
|
status, _ = self._request("POST", APPS_PATH, {"uuid": self.app_uuid}, self.admin)
|
|
self.assertEqual(201, status)
|
|
status, _ = self._request("POST", USERS_PATH, {"user_id": self.user_id}, self.admin)
|
|
self.assertEqual(201, status)
|
|
|
|
def _register(self, agent_id="agent-1", tags=("coreci",)):
|
|
return self._request("POST", REGISTER_PATH, {
|
|
"agent_id": agent_id,
|
|
"app_registration_uuid": self.app_uuid,
|
|
"user_id": self.user_id,
|
|
"tags": list(tags),
|
|
})
|
|
|
|
def _audit(self, action=None, result=None):
|
|
with open(self.env["BROKER_AUDIT_LOG_PATH"], "r", encoding="utf-8") as handle:
|
|
entries = [json.loads(raw) for raw in handle if raw.strip()]
|
|
return [
|
|
entry for entry in entries
|
|
if (action is None or entry["action"] == action)
|
|
and (result is None or entry["result"] == result)
|
|
]
|
|
|
|
def test_agent_id_bare_dot_and_dotdot_rejected(self):
|
|
for agent_id in (".", ".."):
|
|
status, body = self._register(agent_id=agent_id)
|
|
self.assertEqual(400, status, repr(agent_id))
|
|
self.assertEqual(AGENT_ID_BODY, body)
|
|
self.assertEqual([], self.broker.store.agents_by_uuid(self.app_uuid))
|
|
self.assertEqual([], self._audit("register"))
|
|
|
|
def test_agent_id_control_chars_rejected(self):
|
|
for agent_id in ("a\x00b", "a\nb", "a\x1fb", "a\x7fb"):
|
|
status, body = self._register(agent_id=agent_id)
|
|
self.assertEqual(400, status, repr(agent_id))
|
|
self.assertEqual(AGENT_ID_BODY, body)
|
|
self.assertEqual([], self.broker.store.agents_by_uuid(self.app_uuid))
|
|
self.assertEqual([], self._audit("register"))
|
|
|
|
def test_agent_id_embedded_dots_still_accepted(self):
|
|
status, body = self._register(agent_id="a.b")
|
|
self.assertEqual(201, status)
|
|
self.assertTrue(body["created"])
|
|
status, body = self._register(agent_id="tool.v2")
|
|
self.assertEqual(201, status)
|
|
self.assertTrue(body["created"])
|
|
self.assertIsNotNone(self.broker.store.get_agent("a.b", self.app_uuid))
|
|
self.assertIsNotNone(self.broker.store.get_agent("tool.v2", self.app_uuid))
|
|
self.assertEqual(2, len(self._audit("register", "ok")))
|
|
|
|
def test_tag_control_chars_rejected(self):
|
|
for tags in (("ok", "ba\x00d"), ("ok", "ba\x7fd"), ("ok", "ba:d")):
|
|
status, body = self._register(tags=tags)
|
|
self.assertEqual(400, status, repr(tags))
|
|
self.assertEqual(TAGS_BODY, body)
|
|
self.assertEqual([], self.broker.store.agents_by_uuid(self.app_uuid))
|
|
status, body = self._register(tags=("a.b", ".."))
|
|
self.assertEqual(201, status)
|
|
self.assertTrue(body["created"])
|
|
row = self.broker.store.get_agent("agent-1", self.app_uuid)
|
|
self.assertEqual(["..", "a.b"], json.loads(row[3]))
|
|
|
|
def test_agent_id_path_separators_still_rejected(self):
|
|
for agent_id in ("a:b", "a/b", "a\\b"):
|
|
status, body = self._register(agent_id=agent_id)
|
|
self.assertEqual(400, status, repr(agent_id))
|
|
self.assertEqual(AGENT_ID_BODY, body)
|
|
self.assertEqual([], self._audit("register"))
|
|
status, body = self._register(agent_id="plain-agent")
|
|
self.assertEqual(201, status)
|
|
self.assertTrue(body["created"])
|
|
|
|
def test_user_allowlist_add_shares_id_rules(self):
|
|
for user_id in (".", "..", "a\x00b", "a:b", "a/b"):
|
|
status, body = self._request("POST", USERS_PATH, {"user_id": user_id}, self.admin)
|
|
self.assertEqual(400, status, repr(user_id))
|
|
self.assertEqual(USER_ID_BODY, body)
|
|
status, body = self._request("POST", USERS_PATH, {"user_id": "ceo.two"}, self.admin)
|
|
self.assertEqual(201, status)
|
|
|
|
|
|
class P5ReexecGuardTests(unittest.TestCase):
|
|
def setUp(self):
|
|
coreci_broker._reset_server_guard()
|
|
self.addCleanup(coreci_broker._reset_server_guard)
|
|
tmp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(tmp.cleanup)
|
|
self.tmp = tmp.name
|
|
|
|
def _env(self, bind):
|
|
return {
|
|
"BROKER_ENV_PATH": os.path.join(self.tmp, ".env"),
|
|
"BROKER_DB_PATH": os.path.join(self.tmp, "broker.db"),
|
|
"BROKER_AUDIT_LOG_PATH": os.path.join(self.tmp, "broker.audit.log"),
|
|
"BROKER_BIND": bind,
|
|
}
|
|
|
|
def _exec_broker(self):
|
|
with open(coreci_broker.__file__, "r", encoding="utf-8") as handle:
|
|
source = handle.read()
|
|
namespace = {"__name__": "coreci_broker_reexec"}
|
|
exec(compile(source, "coreci_broker.py", "exec"), namespace)
|
|
return namespace
|
|
|
|
def _healthz(self, port):
|
|
url = "http://127.0.0.1:%d/healthz" % port
|
|
with urllib.request.urlopen(url, timeout=5) as response:
|
|
return response.status, json.loads(response.read().decode("utf-8"))
|
|
|
|
def test_fresh_namespace_tools_fails_loud_when_old_module_serves(self):
|
|
first = self._exec_broker()
|
|
with mock.patch.dict(os.environ, self._env("127.0.0.1:0"), clear=True):
|
|
with contextlib.redirect_stdout(io.StringIO()):
|
|
first_tools = first["Tools"]()
|
|
first_again = first["Tools"]()
|
|
server = first["_started_server"]
|
|
self.addCleanup(first["_reset_server_guard"])
|
|
self.assertIs(server, first_tools._server)
|
|
self.assertIs(first_tools._server, first_again._server)
|
|
port = server.server_address[1]
|
|
self.assertNotEqual(0, port)
|
|
second = self._exec_broker()
|
|
self.assertIsNone(second["_started_server"])
|
|
with mock.patch.dict(os.environ, self._env("127.0.0.1:%d" % port), clear=True):
|
|
with contextlib.redirect_stdout(io.StringIO()):
|
|
with self.assertRaises(RuntimeError) as caught:
|
|
second["Tools"]()
|
|
message = str(caught.exception).lower()
|
|
self.assertIn("restart", message)
|
|
self.assertIn("open webui", message)
|
|
self.assertIsInstance(caught.exception.__cause__, OSError)
|
|
self.assertEqual(errno.EADDRINUSE, caught.exception.__cause__.errno)
|
|
self.assertIsNone(second["_started_server"])
|
|
self.assertEqual((200, {"ok": True}), self._healthz(port))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main() |