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---
305 lines
13 KiB
Python
305 lines
13 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.parse
|
|
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"
|
|
AUDIT_PATH = "/api/broker/audit"
|
|
UUID_A = "0b6c3370-6d38-4b92-b9a2-bc2f1c9c1b0a"
|
|
UUID_B = "3f2d8a04-9c17-4b1e-8d2f-6a5b4c3d2e1f"
|
|
|
|
SEED = [
|
|
{"action": "read", "path": "coreci/proxmox", "user_id": "staff-1",
|
|
"app_registration_uuid": UUID_A, "agent_id": "proxmox"},
|
|
{"action": "read", "path": "coreci/gitea", "user_id": "staff-2",
|
|
"app_registration_uuid": UUID_A, "agent_id": "gitea"},
|
|
{"action": "denied", "path": "coreci/gitea", "result": "denied",
|
|
"user_id": "staff-2", "app_registration_uuid": UUID_B, "agent_id": "gitea"},
|
|
{"action": "add_app", "path": "/api/broker/app_registrations"},
|
|
{"action": "read", "path": "coreci/proxmox", "user_id": "staff-1",
|
|
"app_registration_uuid": UUID_B, "agent_id": "proxmox"},
|
|
{"action": "login", "path": "/api/broker/auth/login"},
|
|
{"action": "remove_user", "path": "/api/broker/user_allowlist/staff-2"},
|
|
]
|
|
|
|
|
|
def _basic_header(username, password):
|
|
raw = "%s:%s" % (username, password)
|
|
return "Basic " + base64.b64encode(raw.encode("utf-8")).decode("ascii")
|
|
|
|
|
|
class AuditQueryTests(unittest.TestCase):
|
|
def setUp(self):
|
|
tmp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(tmp.cleanup)
|
|
self.audit_path = os.path.join(tmp.name, "broker.audit.log")
|
|
with contextlib.redirect_stdout(io.StringIO()):
|
|
self.broker = coreci_broker.boot({
|
|
"BROKER_ENV_PATH": os.path.join(tmp.name, ".env"),
|
|
"BROKER_DB_PATH": os.path.join(tmp.name, "broker.db"),
|
|
"BROKER_AUDIT_LOG_PATH": self.audit_path,
|
|
"BROKER_ADMIN_USER": ADMIN_USER,
|
|
"BROKER_ADMIN_PASS": ADMIN_PASS,
|
|
})
|
|
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.auth = {"Authorization": _basic_header(ADMIN_USER, ADMIN_PASS)}
|
|
|
|
def _request_json(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 _get(self, params=None, headers=None):
|
|
path = AUDIT_PATH
|
|
if params:
|
|
path += "?" + urllib.parse.urlencode(params)
|
|
return self._request_json(
|
|
"GET", path, headers=self.auth if headers is None else headers
|
|
)
|
|
|
|
def _seed(self, specs):
|
|
for spec in specs:
|
|
self.broker.audit.write(**spec)
|
|
time.sleep(0.005)
|
|
|
|
def _ts(self):
|
|
values = []
|
|
with open(self.audit_path, "r", encoding="utf-8") as handle:
|
|
for raw in handle:
|
|
try:
|
|
values.append(json.loads(raw)["ts"])
|
|
except (ValueError, TypeError, KeyError):
|
|
continue
|
|
return values
|
|
|
|
def _paths(self, body):
|
|
return [entry["path"] for entry in body["entries"]]
|
|
|
|
def test_no_filters_returns_all_entries_with_null_cursor(self):
|
|
self._seed(SEED)
|
|
status, body = self._get()
|
|
self.assertEqual(200, status)
|
|
with open(self.audit_path, "r", encoding="utf-8") as handle:
|
|
expected = [json.loads(raw) for raw in handle]
|
|
self.assertEqual(expected, body["entries"])
|
|
self.assertEqual(len(SEED), len(body["entries"]))
|
|
self.assertIsNone(body["next_since"])
|
|
|
|
def test_user_id_filter_exact_match(self):
|
|
self._seed(SEED)
|
|
status, body = self._get({"user_id": "staff-1"})
|
|
self.assertEqual(200, status)
|
|
self.assertEqual(["coreci/proxmox", "coreci/proxmox"], self._paths(body))
|
|
self.assertEqual([UUID_A, UUID_B],
|
|
[e["app_registration_uuid"] for e in body["entries"]])
|
|
self.assertIsNone(body["next_since"])
|
|
|
|
def test_app_registration_uuid_filter_exact_match(self):
|
|
self._seed(SEED)
|
|
status, body = self._get({"app_registration_uuid": UUID_B})
|
|
self.assertEqual(200, status)
|
|
self.assertEqual(["denied", "read"], [e["action"] for e in body["entries"]])
|
|
self.assertEqual(["coreci/gitea", "coreci/proxmox"], self._paths(body))
|
|
status, body = self._get({"app_registration_uuid": "no-such-uuid"})
|
|
self.assertEqual(200, status)
|
|
self.assertEqual([], body["entries"])
|
|
self.assertIsNone(body["next_since"])
|
|
|
|
def test_action_filter_exact_match(self):
|
|
self._seed(SEED)
|
|
status, body = self._get({"action": "denied"})
|
|
self.assertEqual(200, status)
|
|
self.assertEqual(["coreci/gitea"], self._paths(body))
|
|
status, body = self._get({"action": "read"})
|
|
self.assertEqual(["coreci/proxmox", "coreci/gitea", "coreci/proxmox"],
|
|
self._paths(body))
|
|
status, body = self._get({"action": "rea"})
|
|
self.assertEqual(200, status)
|
|
self.assertEqual([], body["entries"])
|
|
|
|
def test_since_filter_is_exclusive(self):
|
|
self._seed(SEED)
|
|
ts = self._ts()
|
|
status, body = self._get({"since": ts[2]})
|
|
self.assertEqual(200, status)
|
|
self.assertEqual(["add_app", "read", "login", "remove_user"],
|
|
[e["action"] for e in body["entries"]])
|
|
self.assertEqual(4, len(body["entries"]))
|
|
self.assertIsNone(body["next_since"])
|
|
status, body = self._get({"since": ts[-1]})
|
|
self.assertEqual(200, status)
|
|
self.assertEqual([], body["entries"])
|
|
self.assertIsNone(body["next_since"])
|
|
|
|
def test_until_filter_is_inclusive(self):
|
|
self._seed(SEED)
|
|
ts = self._ts()
|
|
status, body = self._get({"until": ts[2]})
|
|
self.assertEqual(200, status)
|
|
self.assertEqual(["read", "read", "denied"],
|
|
[e["action"] for e in body["entries"]])
|
|
self.assertEqual(3, len(body["entries"]))
|
|
self.assertIsNone(body["next_since"])
|
|
|
|
def test_since_and_until_range_together(self):
|
|
self._seed(SEED)
|
|
ts = self._ts()
|
|
status, body = self._get({"since": ts[1], "until": ts[4]})
|
|
self.assertEqual(200, status)
|
|
self.assertEqual(["denied", "add_app", "read"], [e["action"] for e in body["entries"]])
|
|
status, body = self._get({"since": ts[3], "until": ts[3]})
|
|
self.assertEqual(200, status)
|
|
self.assertEqual([], body["entries"])
|
|
|
|
def test_combined_filters(self):
|
|
self._seed(SEED)
|
|
ts = self._ts()
|
|
status, body = self._get({"user_id": "staff-1", "action": "read"})
|
|
self.assertEqual(200, status)
|
|
self.assertEqual(["coreci/proxmox", "coreci/proxmox"], self._paths(body))
|
|
status, body = self._get({"user_id": "staff-2", "action": "read"})
|
|
self.assertEqual(["coreci/gitea"], self._paths(body))
|
|
status, body = self._get(
|
|
{"app_registration_uuid": UUID_B, "since": ts[2], "until": ts[6]}
|
|
)
|
|
self.assertEqual(["coreci/proxmox"], self._paths(body))
|
|
status, body = self._get(
|
|
{"user_id": "staff-1", "since": ts[0], "until": ts[4]}
|
|
)
|
|
self.assertEqual(["coreci/proxmox"], self._paths(body))
|
|
status, body = self._get(
|
|
{"user_id": "staff-1", "action": "denied",
|
|
"app_registration_uuid": UUID_A, "since": ts[0], "until": ts[6]}
|
|
)
|
|
self.assertEqual([], body["entries"])
|
|
|
|
def test_limit_pagination_with_next_since_cursor(self):
|
|
specs = [{"action": "read", "path": "coreci/agent-%d" % i} for i in range(5)]
|
|
self._seed(specs)
|
|
ts = self._ts()
|
|
status, body = self._get({"limit": 2})
|
|
self.assertEqual(200, status)
|
|
self.assertEqual(["coreci/agent-0", "coreci/agent-1"], self._paths(body))
|
|
self.assertEqual(ts[1], body["next_since"])
|
|
status, body = self._get({"since": body["next_since"], "limit": 2})
|
|
self.assertEqual(200, status)
|
|
self.assertEqual(["coreci/agent-2", "coreci/agent-3"], self._paths(body))
|
|
self.assertEqual(ts[3], body["next_since"])
|
|
status, body = self._get({"since": ts[1]})
|
|
self.assertEqual(200, status)
|
|
self.assertEqual(["coreci/agent-2", "coreci/agent-3", "coreci/agent-4"],
|
|
self._paths(body))
|
|
self.assertIsNone(body["next_since"])
|
|
status, body = self._get({"since": ts[4]})
|
|
self.assertEqual(200, status)
|
|
self.assertEqual([], body["entries"])
|
|
self.assertIsNone(body["next_since"])
|
|
|
|
def test_limit_zero_and_negative_clamp_to_one(self):
|
|
self._seed([{"action": "read", "path": "coreci/a"},
|
|
{"action": "read", "path": "coreci/b"},
|
|
{"action": "read", "path": "coreci/c"}])
|
|
ts = self._ts()
|
|
for bad in (0, -3):
|
|
status, body = self._get({"limit": bad})
|
|
self.assertEqual(200, status)
|
|
self.assertEqual(["coreci/a"], self._paths(body))
|
|
self.assertEqual(ts[0], body["next_since"])
|
|
|
|
def test_limit_above_max_and_non_numeric_are_clamped(self):
|
|
self._seed([{"action": "read", "path": "coreci/a"},
|
|
{"action": "read", "path": "coreci/b"},
|
|
{"action": "read", "path": "coreci/c"}])
|
|
for value in (5000, "abc"):
|
|
status, body = self._get({"limit": value})
|
|
self.assertEqual(200, status)
|
|
self.assertEqual(3, len(body["entries"]))
|
|
self.assertIsNone(body["next_since"])
|
|
|
|
def test_default_limit_is_exactly_100(self):
|
|
specs = [{"action": "read", "path": "coreci/agent-%d" % i} for i in range(100)]
|
|
self._seed(specs)
|
|
ts = self._ts()
|
|
status, body = self._get()
|
|
self.assertEqual(200, status)
|
|
self.assertEqual(100, len(body["entries"]))
|
|
self.assertEqual("coreci/agent-99", body["entries"][-1]["path"])
|
|
self.assertEqual(ts[99], body["next_since"])
|
|
status, body = self._get({"since": ts[99]})
|
|
self.assertEqual(200, status)
|
|
self.assertEqual([], body["entries"])
|
|
self.assertIsNone(body["next_since"])
|
|
|
|
def test_invalid_since_or_until_is_400(self):
|
|
self._seed(SEED)
|
|
status, body = self._get({"since": "not-a-timestamp"})
|
|
self.assertEqual(400, status)
|
|
self.assertEqual(400, body["error"])
|
|
self.assertTrue(body["message"])
|
|
status, body = self._get({"until": "2026-02-30T00:00:00Z"})
|
|
self.assertEqual(400, status)
|
|
self.assertEqual(400, body["error"])
|
|
self.assertTrue(body["message"])
|
|
status, body = self._get({"since": ""})
|
|
self.assertEqual(200, status)
|
|
self.assertEqual(len(SEED), len(body["entries"]))
|
|
|
|
def test_malformed_lines_skipped(self):
|
|
self._seed([{"action": "read", "path": "coreci/a"},
|
|
{"action": "read", "path": "coreci/b"}])
|
|
with open(self.audit_path, "a", encoding="utf-8") as handle:
|
|
handle.write("{not json\n")
|
|
handle.write("[1, 2, 3]\n")
|
|
self._seed([{"action": "read", "path": "coreci/c"}])
|
|
ts = self._ts()
|
|
self.assertEqual(3, len(ts))
|
|
status, body = self._get()
|
|
self.assertEqual(200, status)
|
|
self.assertEqual(["coreci/a", "coreci/b", "coreci/c"], self._paths(body))
|
|
self.assertIsNone(body["next_since"])
|
|
status, body = self._get({"limit": 2})
|
|
self.assertEqual(200, status)
|
|
self.assertEqual(["coreci/a", "coreci/b"], self._paths(body))
|
|
self.assertEqual(ts[1], body["next_since"])
|
|
status, body = self._get({"since": ts[1], "action": "read"})
|
|
self.assertEqual(200, status)
|
|
self.assertEqual(["coreci/c"], self._paths(body))
|
|
|
|
def test_unauthenticated_is_401(self):
|
|
self._seed(SEED)
|
|
status, body = self._get(headers={})
|
|
self.assertEqual(401, status)
|
|
self.assertEqual(401, body["error"])
|
|
status, body = self._get(
|
|
headers={"Authorization": _basic_header(ADMIN_USER, "wrong-pass")}
|
|
)
|
|
self.assertEqual(401, status)
|
|
self.assertEqual(401, body["error"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main() |