broker/tests/test_allowlist.py

import base64
import contextlib
import io
import json
import os
import sys
import tempfile
import unittest
import urllib.error
import urllib.request

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"
BASE_PATH = "/api/broker/user_allowlist"


def _basic_header(username, password):
    raw = "%s:%s" % (username, password)
    return "Basic " + base64.b64encode(raw.encode("utf-8")).decode("ascii")


class UserAllowlistTests(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.server_close)
        self.addCleanup(self.server.shutdown)
        self.base = "http://127.0.0.1:%d" % self.server.server_address[1]
        self.admin = {"Authorization": _basic_header(ADMIN_USER, ADMIN_PASS)}

    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=5) 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 _add(self, user_id, email=None, headers=None):
        payload = {"user_id": user_id}
        if email is not None:
            payload["email"] = email
        return self._request("POST", BASE_PATH, payload, headers or self.admin)

    def _row(self, user_id):
        status, body = self._request("GET", BASE_PATH, headers=self.admin)
        self.assertEqual(200, status)
        for row in body["allowlist"]:
            if row["user_id"] == user_id:
                return row
        return None

    def _audit(self):
        with open(self.env["BROKER_AUDIT_LOG_PATH"], "r", encoding="utf-8") as handle:
            return [json.loads(raw) for raw in handle if raw.strip()]

    def test_list_empty_then_add_then_list(self):
        status, body = self._request("GET", BASE_PATH, headers=self.admin)
        self.assertEqual(200, status)
        self.assertEqual({"allowlist": []}, body)
        status, body = self._add("alice", "alice@coreci.dev")
        self.assertEqual(201, status)
        self.assertEqual({"user_id": "alice", "status": "active"}, body)
        row = self._row("alice")
        self.assertEqual("alice@coreci.dev", row["email"])
        self.assertEqual("active", row["status"])
        self.assertTrue(row["created_at"])
        self.assertIsNone(row["revoked_at"])

    def test_add_without_email_stores_null_email(self):
        status, _ = self._add("bob")
        self.assertEqual(201, status)
        self.assertIsNone(self._row("bob")["email"])
        status, _ = self._request("POST", BASE_PATH, {"user_id": "bob2", "email": 42}, self.admin)
        self.assertEqual(201, status)
        self.assertIsNone(self._row("bob2")["email"])

    def test_add_missing_or_empty_user_id_is_400(self):
        for payload in ({"email": "x@coreci.dev"}, {}, {"user_id": ""},
                        {"user_id": None}, {"user_id": 12345},
                        {"user_id": ["alice"]}, [1, 2]):
            status, body = self._request("POST", BASE_PATH, payload, self.admin)
            self.assertEqual(400, status, repr(payload))
            self.assertEqual(400, body["error"])

    def test_add_duplicate_active_or_revoked_is_409(self):
        self._add("carol", "carol@coreci.dev")
        status, body = self._add("carol")
        self.assertEqual(409, status)
        self.assertEqual(409, body["error"])
        self.assertTrue(body["message"])
        status, _ = self._request("DELETE", BASE_PATH + "/carol", headers=self.admin)
        self.assertEqual(200, status)
        status, body = self._add("carol")
        self.assertEqual(409, status)

    def test_revoke_revoke_again_reinstate_lifecycle(self):
        self._add("dave", "dave@coreci.dev")
        status, body = self._request("DELETE", BASE_PATH + "/dave", headers=self.admin)
        self.assertEqual(200, status)
        self.assertEqual("revoked", body["status"])
        revoked_at = self._row("dave")["revoked_at"]
        self.assertIsNotNone(revoked_at)
        self.assertEqual("revoked", self._row("dave")["status"])
        status, body = self._request("DELETE", BASE_PATH + "/dave", headers=self.admin)
        self.assertEqual(200, status)
        self.assertEqual(revoked_at, self._row("dave")["revoked_at"])
        status, body = self._request(
            "POST", BASE_PATH + "/dave/reinstate", headers=self.admin
        )
        self.assertEqual(200, status)
        self.assertEqual("active", body["status"])
        row = self._row("dave")
        self.assertEqual("active", row["status"])
        self.assertIsNone(row["revoked_at"])

    def test_revoke_missing_is_404(self):
        status, body = self._request("DELETE", BASE_PATH + "/ghost", headers=self.admin)
        self.assertEqual(404, status)
        self.assertEqual(404, body["error"])

    def test_reinstate_missing_is_404(self):
        status, body = self._request(
            "POST", BASE_PATH + "/ghost/reinstate", headers=self.admin
        )
        self.assertEqual(404, status)
        self.assertEqual(404, body["error"])

    def test_admin_actions_audited_with_null_identity(self):
        self._add("erin", "erin@coreci.dev")
        self._request("DELETE", BASE_PATH + "/erin", headers=self.admin)
        self._request("POST", BASE_PATH + "/erin/reinstate", headers=self.admin)
        entries = self._audit()
        self.assertEqual(["add_user", "remove_user", "add_user"],
                         [entry["action"] for entry in entries])
        for entry in entries:
            self.assertEqual("ok", entry["result"])
            self.assertIsNone(entry["user_id"])
            self.assertIsNone(entry["app_registration_uuid"])
            self.assertIsNone(entry["agent_id"])
            self.assertEqual("/api/broker/user_allowlist/erin", entry["path"])

    def test_unauthenticated_requests_are_401(self):
        checks = (
            ("GET", BASE_PATH, None),
            ("POST", BASE_PATH, {"user_id": "zoe"}),
            ("DELETE", BASE_PATH + "/zoe", None),
            ("POST", BASE_PATH + "/zoe/reinstate", None),
        )
        for method, path, payload in checks:
            status, body = self._request(method, path, payload)
            self.assertEqual(401, status, method + " " + path)
            self.assertEqual(401, body["error"])


if __name__ == "__main__":
    unittest.main()