broker/tests/test_admin_auth.py

import base64
import contextlib
import datetime
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"
LOGIN_PATH = "/api/broker/auth/login"
LOGOUT_PATH = "/api/broker/auth/logout"
WHOAMI_PATH = "/api/broker/auth/whoami"


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


def _iso_offset(seconds):
    now = datetime.datetime.now(datetime.timezone.utc)
    return (now + datetime.timedelta(seconds=seconds)).isoformat()


class AdminAuthTests(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,
        }
        self.broker = self._boot_broker(self.env)
        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]

    def _boot_broker(self, env):
        with contextlib.redirect_stdout(io.StringIO()):
            broker = coreci_broker.boot(dict(env))
        self.addCleanup(broker.audit.close)
        self.addCleanup(broker.store.close)
        return broker

    def _request(self, method, path, payload=None, headers=None, base=None):
        data = None if payload is None else json.dumps(payload).encode("utf-8")
        request = urllib.request.Request(
            (base or 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, response.read(), response.headers
        except urllib.error.HTTPError as error:
            return error.code, error.read(), error.headers

    def _request_json(self, method, path, payload=None, headers=None, base=None):
        status, body, response_headers = self._request(
            method, path, payload, headers, base
        )
        return status, json.loads(body.decode("utf-8")), response_headers

    def _login(self, username=ADMIN_USER, password=ADMIN_PASS, base=None):
        return self._request_json(
            "POST",
            LOGIN_PATH,
            {"username": username, "password": password},
            base=base,
        )

    def _login_cookie(self, base=None):
        status, body, headers = self._login(base=base)
        self.assertEqual(200, status)
        self.assertEqual({"ok": True, "username": ADMIN_USER}, body)
        raw = headers.get("Set-Cookie")
        self.assertTrue(raw)
        return raw

    def _cookie_id(self, raw):
        return raw.split(";", 1)[0].split("=", 1)[1]

    def _audit_entries(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 _audit_find(self, action, result):
        for entry in self._audit_entries():
            if entry["action"] == action and entry["result"] == result:
                return entry
        return None

    def _assert_null_identity(self, entry):
        self.assertIsNotNone(entry)
        self.assertIsNone(entry["user_id"])
        self.assertIsNone(entry["app_registration_uuid"])
        self.assertIsNone(entry["agent_id"])

    def test_login_logout_whoami_flow_and_cookie_contract(self):
        raw = self._login_cookie()
        self.assertTrue(raw.startswith("broker_session="))
        self.assertIn("Path=/", raw)
        self.assertIn("HttpOnly", raw)
        self.assertIn("SameSite=Lax", raw)
        cookie = {"Cookie": "broker_session=" + self._cookie_id(raw)}
        status, body, _ = self._request_json("GET", WHOAMI_PATH, headers=cookie)
        self.assertEqual(200, status)
        self.assertEqual(ADMIN_USER, body["username"])
        created = datetime.datetime.fromisoformat(body["created_at"])
        expires = datetime.datetime.fromisoformat(body["expires_at"])
        self.assertEqual(datetime.timedelta(seconds=300), expires - created)
        status, body, _ = self._request_json("POST", LOGOUT_PATH, headers=cookie)
        self.assertEqual(200, status)
        self.assertEqual({"ok": True}, body)
        status, body, _ = self._request_json("GET", WHOAMI_PATH, headers=cookie)
        self.assertEqual(401, status)
        self.assertEqual(401, body["error"])
        self.assertIsNone(self.broker.store.get_admin_session(self._cookie_id(raw)))
        self._assert_null_identity(self._audit_find("login", "ok"))
        self._assert_null_identity(self._audit_find("logout", "ok"))

    def test_wrong_password_is_401_with_denied_audit(self):
        status, body, headers = self._login(password="not-the-pass")
        self.assertEqual(401, status)
        self.assertEqual(401, body["error"])
        self.assertTrue(body["message"])
        self.assertIsNone(headers.get("Set-Cookie"))
        self._assert_null_identity(self._audit_find("login", "denied"))

    def test_wrong_username_is_401(self):
        status, body, headers = self._login(username="mallory")
        self.assertEqual(401, status)
        self.assertIsNone(headers.get("Set-Cookie"))
        self.assertIsNotNone(self._audit_find("login", "denied"))

    def test_login_with_missing_credentials_is_401(self):
        status, body, _ = self._request_json("POST", LOGIN_PATH, payload={})
        self.assertEqual(401, status)
        status, body, _ = self._request_json("POST", LOGIN_PATH)
        self.assertEqual(401, status)
        self.assertIsNotNone(self._audit_find("login", "denied"))

    def test_expired_session_is_rejected_lazily(self):
        raw = self._login_cookie()
        session_id = self._cookie_id(raw)
        self.broker.store.conn.execute(
            "UPDATE admin_sessions SET expires_at = ? WHERE session_id = ?",
            (_iso_offset(-60), session_id),
        )
        self.broker.store.conn.commit()
        status, body, _ = self._request_json(
            "GET", WHOAMI_PATH, headers={"Cookie": "broker_session=" + session_id}
        )
        self.assertEqual(401, status)
        self.assertEqual(401, body["error"])

    def test_boot_cleanup_deletes_stale_sessions_keeps_live(self):
        self.broker.store.conn.executemany(
            "INSERT INTO admin_sessions VALUES (?, ?, ?)",
            [
                ("stale-session", _iso_offset(-600), _iso_offset(-60)),
                ("live-session", _iso_offset(-600), _iso_offset(300)),
            ],
        )
        self.broker.store.conn.commit()
        rebooted = self._boot_broker(self.env)
        self.assertIsNone(rebooted.store.get_admin_session("stale-session"))
        self.assertIsNotNone(rebooted.store.get_admin_session("live-session"))

    def test_basic_valid_on_whoami(self):
        headers = {"Authorization": _basic_header(ADMIN_USER, ADMIN_PASS)}
        status, body, _ = self._request_json("GET", WHOAMI_PATH, headers=headers)
        self.assertEqual(200, status)
        self.assertEqual(ADMIN_USER, body["username"])
        self.assertIsNone(body["created_at"])
        self.assertIsNone(body["expires_at"])

    def test_basic_invalid_is_401(self):
        headers = {"Authorization": _basic_header(ADMIN_USER, "wrong-pass")}
        status, body, _ = self._request_json("GET", WHOAMI_PATH, headers=headers)
        self.assertEqual(401, status)
        self.assertEqual(401, body["error"])
        self.assertTrue(body["message"])

    def test_whoami_without_credentials_is_401(self):
        status, body, _ = self._request_json("GET", WHOAMI_PATH)
        self.assertEqual(401, status)
        self.assertEqual(401, body["error"])

    def test_valid_session_cookie_wins_over_invalid_basic(self):
        raw = self._login_cookie()
        headers = {
            "Cookie": "broker_session=" + self._cookie_id(raw),
            "Authorization": _basic_header(ADMIN_USER, "wrong-pass"),
        }
        status, body, _ = self._request_json("GET", WHOAMI_PATH, headers=headers)
        self.assertEqual(200, status)
        self.assertEqual(ADMIN_USER, body["username"])

    def test_logout_without_credentials_is_401(self):
        status, body, _ = self._request_json("POST", LOGOUT_PATH)
        self.assertEqual(401, status)
        self.assertEqual(401, body["error"])
        self.assertIsNone(self._audit_find("logout", "ok"))

    def test_logout_via_basic_without_session_is_ok(self):
        headers = {"Authorization": _basic_header(ADMIN_USER, ADMIN_PASS)}
        status, body, _ = self._request_json("POST", LOGOUT_PATH, headers=headers)
        self.assertEqual(200, status)
        self.assertEqual({"ok": True}, body)
        self._assert_null_identity(self._audit_find("logout", "ok"))

    def test_session_ttl_is_configurable(self):
        env = dict(self.env)
        env["BROKER_DB_PATH"] = os.path.join(self.tmp, "ttl.db")
        env["BROKER_AUDIT_LOG_PATH"] = os.path.join(self.tmp, "ttl.audit.log")
        env["BROKER_SESSION_TTL"] = "60"
        broker = self._boot_broker(env)
        server = coreci_broker.serve("127.0.0.1", 0, broker)
        self.addCleanup(server.shutdown)
        self.addCleanup(server.server_close)
        base = "http://127.0.0.1:%d" % server.server_address[1]
        raw = self._login_cookie(base=base)
        status, body, _ = self._request_json(
            "GET",
            WHOAMI_PATH,
            headers={"Cookie": "broker_session=" + self._cookie_id(raw)},
            base=base,
        )
        self.assertEqual(200, status)
        created = datetime.datetime.fromisoformat(body["created_at"])
        expires = datetime.datetime.fromisoformat(body["expires_at"])
        self.assertEqual(datetime.timedelta(seconds=60), expires - created)


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