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---
213 lines
7.3 KiB
Python
213 lines
7.3 KiB
Python
import contextlib
|
|
import datetime
|
|
import io
|
|
import json
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
import threading
|
|
import unittest
|
|
|
|
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), os.pardir)))
|
|
|
|
import coreci_broker
|
|
|
|
AUDIT_FIELDS = [
|
|
"ts",
|
|
"user_id",
|
|
"app_registration_uuid",
|
|
"agent_id",
|
|
"action",
|
|
"path",
|
|
"result",
|
|
]
|
|
|
|
ALL_ACTIONS = [
|
|
"register",
|
|
"read",
|
|
"denied",
|
|
"login",
|
|
"logout",
|
|
"add_app",
|
|
"remove_app",
|
|
"add_user",
|
|
"remove_user",
|
|
]
|
|
|
|
ADMIN_ACTIONS = [
|
|
"login",
|
|
"logout",
|
|
"add_app",
|
|
"remove_app",
|
|
"add_user",
|
|
"remove_user",
|
|
]
|
|
|
|
|
|
class AuditTests(unittest.TestCase):
|
|
def setUp(self):
|
|
tmp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(tmp.cleanup)
|
|
self.env_path = os.path.join(tmp.name, ".env")
|
|
self.db_path = os.path.join(tmp.name, "broker.db")
|
|
self.audit_path = os.path.join(tmp.name, "broker.audit.log")
|
|
|
|
def _boot(self, extra_env=None):
|
|
env = {
|
|
"BROKER_ENV_PATH": self.env_path,
|
|
"BROKER_DB_PATH": self.db_path,
|
|
"BROKER_AUDIT_LOG_PATH": self.audit_path,
|
|
}
|
|
if extra_env:
|
|
env.update(extra_env)
|
|
buffer = io.StringIO()
|
|
with contextlib.redirect_stdout(buffer):
|
|
broker = coreci_broker.boot(env)
|
|
self.addCleanup(broker.store.close)
|
|
self.addCleanup(broker.audit.close)
|
|
return broker
|
|
|
|
def _raw_lines(self):
|
|
with open(self.audit_path, "r", encoding="utf-8") as handle:
|
|
return handle.read().splitlines()
|
|
|
|
def _entries(self):
|
|
return [json.loads(raw) for raw in self._raw_lines()]
|
|
|
|
def test_boot_wires_ready_audit_at_mode_0600(self):
|
|
broker = self._boot()
|
|
self.assertIsInstance(broker.audit, coreci_broker.Audit)
|
|
self.assertEqual(self.audit_path, broker.audit.path)
|
|
self.assertTrue(os.path.isfile(self.audit_path))
|
|
self.assertEqual(0o600, os.stat(self.audit_path).st_mode & 0o777)
|
|
|
|
def test_audit_mode_0600_independent_of_umask(self):
|
|
old_umask = os.umask(0o777)
|
|
try:
|
|
broker = self._boot()
|
|
finally:
|
|
os.umask(old_umask)
|
|
broker.audit.write("login", path="/ui/login")
|
|
self.assertEqual(0o600, os.stat(self.audit_path).st_mode & 0o777)
|
|
|
|
def test_each_write_appends_exactly_one_valid_json_line(self):
|
|
broker = self._boot()
|
|
self.assertEqual(0, len(self._raw_lines()))
|
|
broker.audit.write("login", path="/api/broker/auth/login")
|
|
broker.audit.write(
|
|
"read",
|
|
path="coreci/proxmox",
|
|
user_id="staff-7",
|
|
app_registration_uuid="0b6c3370-6d38-4b92-b9a2-bc2f1c9c1b0a",
|
|
agent_id="proxmox",
|
|
)
|
|
broker.audit.write("denied", path="coreci/gitea", result="denied")
|
|
lines = self._raw_lines()
|
|
self.assertEqual(3, len(lines))
|
|
for line in lines:
|
|
self.assertTrue(line)
|
|
entries = self._entries()
|
|
for entry in entries:
|
|
self.assertEqual(AUDIT_FIELDS, list(entry))
|
|
self.assertEqual(["login", "read", "denied"], [e["action"] for e in entries])
|
|
|
|
def test_all_nine_action_verbs_accepted_and_written(self):
|
|
broker = self._boot()
|
|
for action in ALL_ACTIONS:
|
|
broker.audit.write(action, path="/x")
|
|
entries = self._entries()
|
|
self.assertEqual(9, len(entries))
|
|
self.assertEqual(sorted(ALL_ACTIONS), sorted(e["action"] for e in entries))
|
|
|
|
def test_unknown_action_verb_raises_and_appends_nothing(self):
|
|
broker = self._boot()
|
|
for bad in ("bogus", "", "LOGIN", None, "rotate"):
|
|
with self.assertRaises(ValueError):
|
|
broker.audit.write(bad, path="/x")
|
|
with self.assertRaises(ValueError):
|
|
broker.audit.write("read", path="/x", result="bogus")
|
|
self.assertEqual(0, len(self._raw_lines()))
|
|
|
|
def test_admin_actions_carry_null_identity_fields(self):
|
|
broker = self._boot()
|
|
for action in ADMIN_ACTIONS:
|
|
broker.audit.write(action, path="/api/broker/app_registrations")
|
|
entries = self._entries()
|
|
self.assertEqual(len(ADMIN_ACTIONS), len(entries))
|
|
for entry in entries:
|
|
self.assertIsNone(entry["user_id"])
|
|
self.assertIsNone(entry["app_registration_uuid"])
|
|
self.assertIsNone(entry["agent_id"])
|
|
broker.audit.write(
|
|
"register",
|
|
path="/api/broker/agents/register",
|
|
user_id="staff-7",
|
|
app_registration_uuid="0b6c3370-6d38-4b92-b9a2-bc2f1c9c1b0a",
|
|
agent_id="proxmox",
|
|
)
|
|
entry = self._entries()[-1]
|
|
self.assertEqual("staff-7", entry["user_id"])
|
|
self.assertEqual("0b6c3370-6d38-4b92-b9a2-bc2f1c9c1b0a", entry["app_registration_uuid"])
|
|
self.assertEqual("proxmox", entry["agent_id"])
|
|
|
|
def test_append_only_preserves_every_line_across_many_writes(self):
|
|
broker = self._boot()
|
|
for i in range(50):
|
|
broker.audit.write("read", path=f"coreci/agent-{i}")
|
|
entries = self._entries()
|
|
self.assertEqual(50, len(entries))
|
|
self.assertEqual("coreci/agent-0", entries[0]["path"])
|
|
self.assertEqual("coreci/agent-49", entries[-1]["path"])
|
|
self.assertEqual(
|
|
[f"coreci/agent-{i}" for i in range(50)],
|
|
[e["path"] for e in entries],
|
|
)
|
|
|
|
def test_reboot_appends_without_truncating_history(self):
|
|
first = self._boot()
|
|
first.audit.write("login", path="/ui/login")
|
|
first.store.close()
|
|
first.audit.close()
|
|
second = self._boot()
|
|
second.audit.write("logout", path="/ui/logout")
|
|
entries = self._entries()
|
|
self.assertEqual(["login", "logout"], [e["action"] for e in entries])
|
|
self.assertEqual(0o600, os.stat(self.audit_path).st_mode & 0o777)
|
|
|
|
def test_ts_parses_as_iso8601_utc(self):
|
|
broker = self._boot()
|
|
broker.audit.write("read", path="coreci/proxmox", user_id="staff-7")
|
|
for entry in self._entries():
|
|
parsed = datetime.datetime.fromisoformat(entry["ts"])
|
|
self.assertEqual(datetime.timedelta(0), parsed.utcoffset())
|
|
|
|
def test_result_values_written(self):
|
|
broker = self._boot()
|
|
broker.audit.write("read", path="coreci/proxmox", result="ok")
|
|
broker.audit.write("read", path="coreci/gitea", result="denied")
|
|
broker.audit.write("read", path="coreci/boom", result="error")
|
|
self.assertEqual(["ok", "denied", "error"], [e["result"] for e in self._entries()])
|
|
|
|
def test_concurrent_writes_are_serialized(self):
|
|
broker = self._boot()
|
|
threads = []
|
|
for t in range(4):
|
|
def worker(t=t):
|
|
for i in range(10):
|
|
broker.audit.write("read", path=f"coreci/t{t}-i{i}")
|
|
thread = threading.Thread(target=worker)
|
|
threads.append(thread)
|
|
for thread in threads:
|
|
thread.start()
|
|
for thread in threads:
|
|
thread.join()
|
|
entries = self._entries()
|
|
self.assertEqual(40, len(entries))
|
|
self.assertEqual(
|
|
sorted(f"coreci/t{t}-i{i}" for t in range(4) for i in range(10)),
|
|
sorted(e["path"] for e in entries),
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main() |