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---
249 lines
11 KiB
Python
249 lines
11 KiB
Python
import base64
|
|
import contextlib
|
|
import datetime
|
|
import io
|
|
import json
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
import urllib.error
|
|
import urllib.request
|
|
import uuid as uuidlib
|
|
|
|
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"
|
|
REGISTER_PATH = "/api/broker/agents/register"
|
|
APPS_PATH = "/api/broker/app_registrations"
|
|
USERS_PATH = "/api/broker/user_allowlist"
|
|
PROTECTED_PATH = "/test/p3-remediation"
|
|
AUTH_401_BODY = {"error": 401, "message": "invalid app credentials"}
|
|
CAP_BODY = {"error": 400, "message": "too many agents for this app"}
|
|
AGENT_ID_BODY = {"error": 400, "message": "invalid agent_id"}
|
|
|
|
|
|
def _basic_header(username, password):
|
|
raw = "%s:%s" % (username, password)
|
|
return "Basic " + base64.b64encode(raw.encode("utf-8")).decode("ascii")
|
|
|
|
|
|
def _protected_echo(request):
|
|
request.send_json(200, dict(request.app_auth))
|
|
|
|
|
|
class P3RemediationTests(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.shutdown)
|
|
self.addCleanup(self.server.server_close)
|
|
self.base = "http://127.0.0.1:%d" % self.server.server_address[1]
|
|
self.admin = {"Authorization": _basic_header(ADMIN_USER, ADMIN_PASS)}
|
|
self.app_uuid = str(uuidlib.uuid4())
|
|
self.user_id = "ceo"
|
|
self._seed(self.app_uuid, self.user_id)
|
|
self.broker.router.add(
|
|
"GET", PROTECTED_PATH, coreci_broker._require_app(_protected_echo)
|
|
)
|
|
|
|
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=10) 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 _seed(self, app_uuid, user_id, base=None):
|
|
status, _ = self._request("POST", APPS_PATH, {"uuid": app_uuid}, self.admin, base)
|
|
self.assertEqual(201, status)
|
|
status, _ = self._request("POST", USERS_PATH, {"user_id": user_id}, self.admin, base)
|
|
self.assertEqual(201, status)
|
|
|
|
def _register(self, agent_id="agent-1", tags=("coreci",), app_uuid=None,
|
|
user_id=None, base=None):
|
|
payload = {
|
|
"agent_id": agent_id,
|
|
"app_registration_uuid": app_uuid or self.app_uuid,
|
|
"user_id": user_id or self.user_id,
|
|
"tags": list(tags),
|
|
}
|
|
return self._request("POST", REGISTER_PATH, payload, base=base)
|
|
|
|
def _auth(self, token, app_uuid=None):
|
|
header = "App %s %s" % (app_uuid or self.app_uuid, token)
|
|
return self._request("GET", PROTECTED_PATH, headers={"Authorization": header})
|
|
|
|
def _audit(self, action=None, result=None):
|
|
with open(self.env["BROKER_AUDIT_LOG_PATH"], "r", encoding="utf-8") as handle:
|
|
entries = [json.loads(raw) for raw in handle if raw.strip()]
|
|
return [
|
|
entry for entry in entries
|
|
if (action is None or entry["action"] == action)
|
|
and (result is None or entry["result"] == result)
|
|
]
|
|
|
|
def _future(self, seconds=3600):
|
|
return (datetime.datetime.now(datetime.timezone.utc)
|
|
+ datetime.timedelta(seconds=seconds)).isoformat()
|
|
|
|
def test_agent_id_colon_slash_backslash_rejected(self):
|
|
for bad in ("a:b", "a/b", "a\\b"):
|
|
status, body = self._register(agent_id=bad)
|
|
self.assertEqual(400, status, repr(bad))
|
|
self.assertEqual(AGENT_ID_BODY, body)
|
|
status, body = self._register(agent_id="a:b", app_uuid=str(uuidlib.uuid4()))
|
|
self.assertEqual(400, status)
|
|
self.assertEqual(AGENT_ID_BODY, body)
|
|
self.assertEqual([], self._audit("register"))
|
|
self.assertEqual([], self.broker.store.agents_by_uuid(self.app_uuid))
|
|
|
|
def test_tag_containing_colon_rejected(self):
|
|
status, body = self._register(tags=("ok", "ba:d"))
|
|
self.assertEqual(400, status)
|
|
self.assertEqual({"error": 400, "message": "invalid tags"}, body)
|
|
self.assertEqual([], self.broker.store.agents_by_uuid(self.app_uuid))
|
|
status, body = self._register(tags=("ok",))
|
|
self.assertEqual(201, status)
|
|
self.assertTrue(body["created"])
|
|
|
|
def test_user_id_colon_rejected_at_allowlist_add(self):
|
|
status, body = self._request("POST", USERS_PATH, {"user_id": "co:o"}, self.admin)
|
|
self.assertEqual(400, status)
|
|
self.assertEqual({"error": 400, "message": "invalid user_id"}, body)
|
|
status, body = self._register(user_id="co:o")
|
|
self.assertEqual(403, status)
|
|
self.assertEqual({"error": 403, "message": "registration not allowed"}, body)
|
|
self.assertEqual(1, len(self._audit("register", "denied")))
|
|
status, body = self._request("POST", USERS_PATH, {"user_id": "coo"}, self.admin)
|
|
self.assertEqual(201, status)
|
|
status, body = self._register(agent_id="coo-agent", user_id="coo")
|
|
self.assertEqual(201, status)
|
|
|
|
def test_default_fifty_agent_cap_blocks_register(self):
|
|
self.assertEqual("50", self.broker.config["BROKER_MAX_AGENTS_PER_APP"])
|
|
now = datetime.datetime.now(datetime.timezone.utc).isoformat()
|
|
seed_hash = coreci_broker.hash_token("seed")
|
|
for index in range(50):
|
|
self.broker.store.save_agent(
|
|
"seeded-%d" % index, self.app_uuid, self.user_id, "[]",
|
|
seed_hash, self._future(), now,
|
|
)
|
|
status, body = self._register(agent_id="overflow")
|
|
self.assertEqual(400, status)
|
|
self.assertEqual(CAP_BODY, body)
|
|
status, body = self._register(agent_id="seeded-0")
|
|
self.assertEqual(400, status)
|
|
self.assertEqual(CAP_BODY, body)
|
|
self.assertIsNone(self.broker.store.get_agent("overflow", self.app_uuid))
|
|
self.assertEqual(50, len(self.broker.store.agents_by_uuid(self.app_uuid)))
|
|
self.assertEqual([], self._audit("register"))
|
|
|
|
def test_max_agents_per_app_env_override_honored(self):
|
|
env = dict(self.env)
|
|
env["BROKER_DB_PATH"] = os.path.join(self.tmp, "cap.db")
|
|
env["BROKER_AUDIT_LOG_PATH"] = os.path.join(self.tmp, "cap.audit.log")
|
|
env["BROKER_MAX_AGENTS_PER_APP"] = "2"
|
|
with contextlib.redirect_stdout(io.StringIO()):
|
|
broker = coreci_broker.boot(dict(env))
|
|
self.addCleanup(broker.audit.close)
|
|
self.addCleanup(broker.store.close)
|
|
self.assertEqual("2", broker.config["BROKER_MAX_AGENTS_PER_APP"])
|
|
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]
|
|
self._seed(self.app_uuid, self.user_id, base=base)
|
|
for agent_id in ("agent-a", "agent-b"):
|
|
status, body = self._register(agent_id=agent_id, base=base)
|
|
self.assertEqual(201, status, agent_id)
|
|
self.assertTrue(body["created"])
|
|
status, body = self._register(agent_id="agent-c", base=base)
|
|
self.assertEqual(400, status)
|
|
self.assertEqual(CAP_BODY, body)
|
|
self.assertIsNone(broker.store.get_agent("agent-c", self.app_uuid))
|
|
self.assertEqual(2, len(broker.store.agents_by_uuid(self.app_uuid)))
|
|
|
|
def test_overwrite_evicts_uuid_cache_entries(self):
|
|
_, first = self._register(tags=("red",))
|
|
old_token = first["token"]
|
|
status, _ = self._auth(old_token)
|
|
self.assertEqual(200, status)
|
|
self.assertIn((self.app_uuid, old_token), self.broker.token_cache)
|
|
status, second = self._register(tags=("blue",))
|
|
self.assertEqual(200, status)
|
|
self.assertFalse(second["created"])
|
|
new_token = second["token"]
|
|
self.assertNotEqual(old_token, new_token)
|
|
self.assertEqual([], [k for k in self.broker.token_cache if k[0] == self.app_uuid])
|
|
status, body = self._auth(old_token)
|
|
self.assertEqual(401, status)
|
|
self.assertEqual(AUTH_401_BODY, body)
|
|
self.assertNotIn((self.app_uuid, old_token), self.broker.token_cache)
|
|
status, body = self._auth(new_token)
|
|
self.assertEqual(200, status)
|
|
self.assertEqual("agent-1", body["agent_id"])
|
|
self.assertIn((self.app_uuid, new_token), self.broker.token_cache)
|
|
row = self.broker.store.get_agent("agent-1", self.app_uuid)
|
|
self.assertFalse(coreci_broker.verify_token(old_token, row[4]))
|
|
self.assertTrue(coreci_broker.verify_token(new_token, row[4]))
|
|
|
|
def test_integrity_error_race_takes_idempotent_path(self):
|
|
agent_id = "race-agent"
|
|
tags = ["coreci"]
|
|
pepper = self.broker.store.get_config("BROKER_INSTANCE_PEPPER")
|
|
token = coreci_broker.derive_token(pepper, self.app_uuid, self.user_id, agent_id, tags)
|
|
expires = self._future()
|
|
now = datetime.datetime.now(datetime.timezone.utc).isoformat()
|
|
self.broker.store.save_agent(
|
|
agent_id, self.app_uuid, self.user_id, json.dumps(sorted(tags)),
|
|
coreci_broker.hash_token(token), expires, now,
|
|
)
|
|
real_get_agent = self.broker.store.get_agent
|
|
calls = []
|
|
|
|
def racing_get_agent(agent_id, app_uuid):
|
|
calls.append(agent_id)
|
|
if len(calls) == 1:
|
|
return None
|
|
return real_get_agent(agent_id, app_uuid)
|
|
|
|
self.broker.store.get_agent = racing_get_agent
|
|
self.addCleanup(setattr, self.broker.store, "get_agent", real_get_agent)
|
|
status, body = self._register(agent_id=agent_id, tags=tags)
|
|
self.assertEqual(200, status)
|
|
self.assertFalse(body["created"])
|
|
self.assertEqual(token, body["token"])
|
|
self.assertEqual(expires, body["token_expires_at"])
|
|
self.assertIn("pepper", body)
|
|
self.assertEqual(2, len(calls))
|
|
row = real_get_agent(agent_id, self.app_uuid)
|
|
self.assertTrue(coreci_broker.verify_token(token, row[4]))
|
|
self.assertGreaterEqual(row[7], now)
|
|
self.assertEqual(1, len(self._audit("register", "ok")))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main() |