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---
182 lines
7.0 KiB
Python
182 lines
7.0 KiB
Python
import base64
|
|
import hashlib
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
|
|
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), os.pardir)))
|
|
|
|
import coreci_broker
|
|
|
|
PEPPER = "ZpQ1hP3rLmX7wYc2vB4nT6kS9dF5gR8j"
|
|
APP_UUID = "3f2a9c1e-77d0-4f0a-9b2e-6c1d5a8e4b3c"
|
|
USER_ID = "user-42"
|
|
AGENT_ID = "proxmox"
|
|
|
|
B64_ALPHABET = set(
|
|
"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/="
|
|
)
|
|
|
|
|
|
def _independent_token(pepper, app_uuid, user_id, agent_id, tags):
|
|
payload = "{}:{}:{}:{}:{}".format(
|
|
pepper,
|
|
app_uuid,
|
|
user_id,
|
|
agent_id,
|
|
",".join(sorted(tags)),
|
|
)
|
|
digest = hashlib.sha256(payload.encode("utf-8")).digest()
|
|
return base64.b64encode(digest).decode("ascii")
|
|
|
|
|
|
class FormulaTests(unittest.TestCase):
|
|
def test_matches_independent_computation(self):
|
|
token = coreci_broker.derive_token(
|
|
PEPPER, APP_UUID, USER_ID, AGENT_ID, ["infra", "proxmox"]
|
|
)
|
|
expected = _independent_token(
|
|
PEPPER, APP_UUID, USER_ID, AGENT_ID, ["infra", "proxmox"]
|
|
)
|
|
self.assertEqual(expected, token)
|
|
|
|
def test_exact_spec_formula_byte_for_byte(self):
|
|
expected = base64.b64encode(
|
|
hashlib.sha256(
|
|
"{}:{}:{}:{}:{}".format(
|
|
PEPPER, APP_UUID, USER_ID, AGENT_ID, "a,b"
|
|
).encode("utf-8")
|
|
).digest()
|
|
).decode("ascii")
|
|
actual = coreci_broker.derive_token(PEPPER, APP_UUID, USER_ID, AGENT_ID, ["a", "b"])
|
|
self.assertEqual(expected, actual)
|
|
|
|
def test_empty_tags_match_independent_computation(self):
|
|
token = coreci_broker.derive_token(PEPPER, APP_UUID, USER_ID, AGENT_ID, [])
|
|
self.assertEqual(
|
|
_independent_token(PEPPER, APP_UUID, USER_ID, AGENT_ID, []),
|
|
token,
|
|
)
|
|
|
|
def test_deterministic_for_same_inputs(self):
|
|
first = coreci_broker.derive_token(PEPPER, APP_UUID, USER_ID, AGENT_ID, ["a"])
|
|
second = coreci_broker.derive_token(PEPPER, APP_UUID, USER_ID, AGENT_ID, ["a"])
|
|
self.assertEqual(first, second)
|
|
|
|
|
|
class ShapeTests(unittest.TestCase):
|
|
def test_token_is_exactly_44_chars(self):
|
|
token = coreci_broker.derive_token(PEPPER, APP_UUID, USER_ID, AGENT_ID, ["a"])
|
|
self.assertEqual(44, len(token))
|
|
|
|
def test_token_is_pure_ascii(self):
|
|
token = coreci_broker.derive_token(PEPPER, APP_UUID, USER_ID, AGENT_ID, ["a"])
|
|
self.assertTrue(token.isascii())
|
|
token.encode("ascii")
|
|
|
|
def test_token_carries_base64_padding_char(self):
|
|
token = coreci_broker.derive_token(PEPPER, APP_UUID, USER_ID, AGENT_ID, ["a"])
|
|
self.assertTrue(token.endswith("="))
|
|
self.assertEqual(1, token.count("="))
|
|
|
|
def test_token_uses_only_base64_alphabet(self):
|
|
token = coreci_broker.derive_token(PEPPER, APP_UUID, USER_ID, AGENT_ID, ["a"])
|
|
self.assertTrue(set(token) <= B64_ALPHABET)
|
|
|
|
def test_token_decodes_to_32_byte_sha256_digest(self):
|
|
token = coreci_broker.derive_token(PEPPER, APP_UUID, USER_ID, AGENT_ID, ["a"])
|
|
self.assertEqual(32, len(base64.b64decode(token)))
|
|
|
|
def test_token_has_no_prefix(self):
|
|
token = coreci_broker.derive_token(PEPPER, APP_UUID, USER_ID, AGENT_ID, ["a"])
|
|
for prefix in ("App", "Bearer", "scrypt$", "$2", "token", "tk_", "coreci"):
|
|
with self.subTest(prefix=prefix):
|
|
self.assertFalse(token.startswith(prefix))
|
|
self.assertNotIn(" ", token)
|
|
self.assertNotIn(":", token)
|
|
self.assertNotIn("$", token)
|
|
|
|
|
|
class TagNormalizationTests(unittest.TestCase):
|
|
def test_tag_order_does_not_change_token(self):
|
|
first = coreci_broker.derive_token(PEPPER, APP_UUID, USER_ID, AGENT_ID, ["b", "a"])
|
|
second = coreci_broker.derive_token(PEPPER, APP_UUID, USER_ID, AGENT_ID, ["a", "b"])
|
|
self.assertEqual(first, second)
|
|
|
|
def test_accepts_tuple_and_set_iterables(self):
|
|
expected = coreci_broker.derive_token(PEPPER, APP_UUID, USER_ID, AGENT_ID, ["a", "b"])
|
|
self.assertEqual(
|
|
expected,
|
|
coreci_broker.derive_token(PEPPER, APP_UUID, USER_ID, AGENT_ID, ("a", "b")),
|
|
)
|
|
self.assertEqual(
|
|
expected,
|
|
coreci_broker.derive_token(PEPPER, APP_UUID, USER_ID, AGENT_ID, {"a", "b"}),
|
|
)
|
|
|
|
|
|
class InputSensitivityTests(unittest.TestCase):
|
|
def test_pepper_changes_token(self):
|
|
base = coreci_broker.derive_token(PEPPER, APP_UUID, USER_ID, AGENT_ID, ["a"])
|
|
other = coreci_broker.derive_token("other-pepper", APP_UUID, USER_ID, AGENT_ID, ["a"])
|
|
self.assertNotEqual(base, other)
|
|
|
|
def test_app_uuid_changes_token(self):
|
|
base = coreci_broker.derive_token(PEPPER, APP_UUID, USER_ID, AGENT_ID, ["a"])
|
|
other = coreci_broker.derive_token(
|
|
PEPPER, "00000000-0000-0000-0000-000000000000", USER_ID, AGENT_ID, ["a"]
|
|
)
|
|
self.assertNotEqual(base, other)
|
|
|
|
def test_user_id_changes_token(self):
|
|
base = coreci_broker.derive_token(PEPPER, APP_UUID, USER_ID, AGENT_ID, ["a"])
|
|
other = coreci_broker.derive_token(PEPPER, APP_UUID, "user-99", AGENT_ID, ["a"])
|
|
self.assertNotEqual(base, other)
|
|
|
|
def test_agent_id_changes_token(self):
|
|
base = coreci_broker.derive_token(PEPPER, APP_UUID, USER_ID, AGENT_ID, ["a"])
|
|
other = coreci_broker.derive_token(PEPPER, APP_UUID, USER_ID, "gitea", ["a"])
|
|
self.assertNotEqual(base, other)
|
|
|
|
def test_tags_change_token(self):
|
|
base = coreci_broker.derive_token(PEPPER, APP_UUID, USER_ID, AGENT_ID, ["a"])
|
|
other = coreci_broker.derive_token(PEPPER, APP_UUID, USER_ID, AGENT_ID, ["a", "c"])
|
|
self.assertNotEqual(base, other)
|
|
|
|
|
|
class PurityTests(unittest.TestCase):
|
|
def test_derivation_writes_nothing_to_disk(self):
|
|
cwd = os.getcwd()
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
os.chdir(tmp)
|
|
try:
|
|
coreci_broker.derive_token(PEPPER, APP_UUID, USER_ID, AGENT_ID, ["a"])
|
|
self.assertEqual([], os.listdir(tmp))
|
|
finally:
|
|
os.chdir(cwd)
|
|
|
|
def test_derivation_isolated_from_store_and_audit(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
db_path = os.path.join(tmp, "broker.db")
|
|
audit_path = os.path.join(tmp, "broker.audit.log")
|
|
store = coreci_broker.Store(db_path)
|
|
audit = coreci_broker.Audit(audit_path)
|
|
try:
|
|
token = coreci_broker.derive_token(
|
|
PEPPER, APP_UUID, USER_ID, AGENT_ID, ["a"]
|
|
)
|
|
rows = store.conn.execute(
|
|
"SELECT * FROM agent_registrations"
|
|
).fetchall()
|
|
self.assertEqual([], rows)
|
|
finally:
|
|
store.close()
|
|
audit.close()
|
|
with open(audit_path, "r", encoding="utf-8") as handle:
|
|
self.assertEqual([], handle.readlines())
|
|
self.assertNotIn(token, open(db_path, "rb").read().decode("latin-1"))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main() |