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---
241 lines
9.8 KiB
Python
241 lines
9.8 KiB
Python
import base64
|
|
import contextlib
|
|
import io
|
|
import json
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
import unittest
|
|
import urllib.error
|
|
import urllib.request
|
|
import uuid as uuidlib
|
|
|
|
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), os.pardir)))
|
|
|
|
import coreci_broker
|
|
|
|
ADMIN_USER = "admin"
|
|
LOGIN_PATH = "/api/broker/auth/login"
|
|
LOGOUT_PATH = "/api/broker/auth/logout"
|
|
WHOAMI_PATH = "/api/broker/auth/whoami"
|
|
APPS_PATH = "/api/broker/app_registrations"
|
|
USERS_PATH = "/api/broker/user_allowlist"
|
|
AUDIT_PATH = "/api/broker/audit"
|
|
|
|
|
|
class AcceptanceP2Tests(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"),
|
|
}
|
|
|
|
def _boot_served(self, extra=None):
|
|
env = dict(self.env)
|
|
env.update(extra or {})
|
|
buffer = io.StringIO()
|
|
with contextlib.redirect_stdout(buffer):
|
|
broker = coreci_broker.boot(env)
|
|
self.addCleanup(broker.audit.close)
|
|
self.addCleanup(broker.store.close)
|
|
printed = [
|
|
line.split("=", 1)[1]
|
|
for line in buffer.getvalue().splitlines()
|
|
if line.startswith("BROKER_ADMIN_PASS=")
|
|
]
|
|
self.assertEqual(1, len(printed))
|
|
server = coreci_broker.serve("127.0.0.1", 0, broker)
|
|
self.addCleanup(server.server_close)
|
|
self.addCleanup(server.shutdown)
|
|
return "http://127.0.0.1:%d" % server.server_address[1], printed[0]
|
|
|
|
def _request(self, base, method, path, payload=None, headers=None):
|
|
data = None if payload is None else json.dumps(payload).encode("utf-8")
|
|
request = urllib.request.Request(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=5) as response:
|
|
return response.status, response.read(), response.headers
|
|
except urllib.error.HTTPError as error:
|
|
return error.code, error.read(), error.headers
|
|
|
|
def _request_json(self, base, method, path, payload=None, headers=None):
|
|
status, body, response_headers = self._request(
|
|
base, method, path, payload, headers
|
|
)
|
|
return status, json.loads(body.decode("utf-8")), response_headers
|
|
|
|
def _get_page(self, base, path):
|
|
status, body, headers = self._request(base, "GET", path)
|
|
self.assertEqual(200, status)
|
|
self.assertTrue(headers.get("Content-Type", "").startswith("text/html"))
|
|
return body.decode("utf-8")
|
|
|
|
def _basic(self, password):
|
|
raw = "%s:%s" % (ADMIN_USER, password)
|
|
encoded = base64.b64encode(raw.encode("utf-8")).decode("ascii")
|
|
return {"Authorization": "Basic " + encoded}
|
|
|
|
def _login(self, base, password):
|
|
status, body, headers = self._request_json(
|
|
base, "POST", LOGIN_PATH, {"username": ADMIN_USER, "password": password}
|
|
)
|
|
self.assertEqual(200, status)
|
|
self.assertEqual({"ok": True, "username": ADMIN_USER}, body)
|
|
raw = headers.get("Set-Cookie")
|
|
self.assertTrue(raw)
|
|
self.assertTrue(raw.startswith("broker_session="))
|
|
self.assertIn("HttpOnly", raw)
|
|
self.assertIn("SameSite=Lax", raw)
|
|
return {"Cookie": raw.split(";", 1)[0]}
|
|
|
|
def _find(self, rows, key, value):
|
|
for row in rows:
|
|
if row[key] == value:
|
|
return row
|
|
self.fail("no row with %s=%r" % (key, value))
|
|
|
|
def _ok_entries(self, entries):
|
|
return [entry for entry in entries if entry["result"] == "ok"]
|
|
|
|
def _assert_null_identity(self, entries):
|
|
self.assertTrue(entries)
|
|
for entry in entries:
|
|
self.assertIsNone(entry["user_id"])
|
|
self.assertIsNone(entry["app_registration_uuid"])
|
|
self.assertIsNone(entry["agent_id"])
|
|
|
|
def test_ac1_first_boot_printed_pass_login_via_ui_login_page(self):
|
|
base, printed_pass = self._boot_served()
|
|
self._get_page(base, "/ui/login")
|
|
cookie = self._login(base, printed_pass)
|
|
status, body, _ = self._request_json(
|
|
base, "GET", WHOAMI_PATH, headers=cookie
|
|
)
|
|
self.assertEqual(200, status)
|
|
self.assertEqual(ADMIN_USER, body["username"])
|
|
|
|
def test_ac2_app_registration_lifecycle_via_session_cookie(self):
|
|
base, printed_pass = self._boot_served()
|
|
cookie = self._login(base, printed_pass)
|
|
app_uuid = str(uuidlib.uuid4())
|
|
status, body, _ = self._request_json(
|
|
base, "POST", APPS_PATH,
|
|
{"uuid": app_uuid, "name": "proxmox-tool"}, headers=cookie,
|
|
)
|
|
self.assertEqual(201, status)
|
|
self.assertEqual({"uuid": app_uuid, "status": "active"}, body)
|
|
status, body, _ = self._request_json(base, "GET", APPS_PATH, headers=cookie)
|
|
self.assertEqual(200, status)
|
|
row = self._find(body["registrations"], "uuid", app_uuid)
|
|
self.assertEqual("proxmox-tool", row["name"])
|
|
self.assertEqual("active", row["status"])
|
|
self._get_page(base, "/ui/app_registrations")
|
|
status, body, _ = self._request_json(
|
|
base, "DELETE", APPS_PATH + "/" + app_uuid, headers=cookie
|
|
)
|
|
self.assertEqual(200, status)
|
|
self.assertEqual("revoked", body["status"])
|
|
status, body, _ = self._request_json(base, "GET", APPS_PATH, headers=cookie)
|
|
self.assertEqual(
|
|
"revoked", self._find(body["registrations"], "uuid", app_uuid)["status"]
|
|
)
|
|
status, body, _ = self._request_json(
|
|
base, "POST", APPS_PATH + "/" + app_uuid + "/reinstate", headers=cookie
|
|
)
|
|
self.assertEqual(200, status)
|
|
self.assertEqual("active", body["status"])
|
|
status, body, _ = self._request_json(base, "GET", APPS_PATH, headers=cookie)
|
|
self.assertEqual(
|
|
"active", self._find(body["registrations"], "uuid", app_uuid)["status"]
|
|
)
|
|
|
|
def test_ac3_user_allowlist_lifecycle_via_session_cookie(self):
|
|
base, printed_pass = self._boot_served()
|
|
cookie = self._login(base, printed_pass)
|
|
status, body, _ = self._request_json(
|
|
base, "POST", USERS_PATH,
|
|
{"user_id": "ceo", "email": "ceo@coreci.dev"}, headers=cookie,
|
|
)
|
|
self.assertEqual(201, status)
|
|
self.assertEqual({"user_id": "ceo", "status": "active"}, body)
|
|
status, body, _ = self._request_json(base, "GET", USERS_PATH, headers=cookie)
|
|
self.assertEqual(200, status)
|
|
row = self._find(body["allowlist"], "user_id", "ceo")
|
|
self.assertEqual("ceo@coreci.dev", row["email"])
|
|
self.assertEqual("active", row["status"])
|
|
self._get_page(base, "/ui/user_allowlist")
|
|
status, body, _ = self._request_json(
|
|
base, "DELETE", USERS_PATH + "/ceo", headers=cookie
|
|
)
|
|
self.assertEqual(200, status)
|
|
self.assertEqual("revoked", body["status"])
|
|
status, body, _ = self._request_json(base, "GET", USERS_PATH, headers=cookie)
|
|
self.assertEqual(
|
|
"revoked", self._find(body["allowlist"], "user_id", "ceo")["status"]
|
|
)
|
|
status, body, _ = self._request_json(
|
|
base, "POST", USERS_PATH + "/ceo/reinstate", headers=cookie
|
|
)
|
|
self.assertEqual(200, status)
|
|
self.assertEqual("active", body["status"])
|
|
status, body, _ = self._request_json(base, "GET", USERS_PATH, headers=cookie)
|
|
self.assertEqual(
|
|
"active", self._find(body["allowlist"], "user_id", "ceo")["status"]
|
|
)
|
|
|
|
def test_ac9_login_and_logout_audit_lines_with_null_identity(self):
|
|
base, printed_pass = self._boot_served()
|
|
cookie = self._login(base, printed_pass)
|
|
status, body, _ = self._request_json(
|
|
base, "POST", LOGOUT_PATH, headers=cookie
|
|
)
|
|
self.assertEqual(200, status)
|
|
self.assertEqual({"ok": True}, body)
|
|
admin = self._basic(printed_pass)
|
|
status, body, _ = self._request_json(
|
|
base, "GET", AUDIT_PATH + "?action=login", headers=admin
|
|
)
|
|
self.assertEqual(200, status)
|
|
logins = self._ok_entries(body["entries"])
|
|
self.assertEqual(["login"], [entry["action"] for entry in logins])
|
|
self._assert_null_identity(logins)
|
|
status, body, _ = self._request_json(
|
|
base, "GET", AUDIT_PATH + "?action=logout", headers=admin
|
|
)
|
|
self.assertEqual(200, status)
|
|
logouts = self._ok_entries(body["entries"])
|
|
self.assertEqual(["logout"], [entry["action"] for entry in logouts])
|
|
self._assert_null_identity(logouts)
|
|
|
|
def test_ui_pages_served_and_static_hygiene(self):
|
|
base, printed_pass = self._boot_served()
|
|
dashboard = self._get_page(base, "/ui/dashboard")
|
|
self.assertIn("Dashboard", dashboard)
|
|
audit_page = self._get_page(base, "/ui/audit")
|
|
self.assertNotIn("export", audit_page.lower())
|
|
login_page = self._get_page(base, "/ui/login")
|
|
self.assertIn('name="username"', login_page)
|
|
self.assertIn('name="password"', login_page)
|
|
|
|
def test_session_ttl_expiry_end_to_end(self):
|
|
base, printed_pass = self._boot_served({"BROKER_SESSION_TTL": "1"})
|
|
cookie = self._login(base, printed_pass)
|
|
status, _, _ = self._request(base, "GET", WHOAMI_PATH, headers=cookie)
|
|
self.assertEqual(200, status)
|
|
time.sleep(1.2)
|
|
status, body, _ = self._request_json(
|
|
base, "GET", WHOAMI_PATH, headers=cookie
|
|
)
|
|
self.assertEqual(401, status)
|
|
self.assertEqual(401, body["error"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main() |