318 lines
16 KiB
Python
318 lines
16 KiB
Python
import copy
|
|
import importlib.util
|
|
import json
|
|
from pathlib import Path
|
|
import tempfile
|
|
import unittest
|
|
|
|
|
|
PACKAGE = Path(__file__).resolve().parents[1]
|
|
spec = importlib.util.spec_from_file_location("prepare_seq_key", PACKAGE / "prepare-seq-key.py")
|
|
seq = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(seq)
|
|
|
|
|
|
def key(token="a" * 64):
|
|
return {"Id": "apikey-notification", "Title": seq.TITLE, "Token": token,
|
|
"AssignedPermissions": ["Ingest"], "InputSettings": {"AppliedProperties": copy.deepcopy(seq.PROPERTIES)}}
|
|
|
|
|
|
class FakeSeq:
|
|
def __init__(self, keys=None, token_file=None, fail_after_create=False, fail_before_put=False, fail_after_put=False):
|
|
self.keys = copy.deepcopy(keys or [])
|
|
self.calls = []
|
|
self.token_file = token_file
|
|
self.fail_after_create = fail_after_create
|
|
self.events = []
|
|
self.fail_before_put = fail_before_put
|
|
self.fail_after_put = fail_after_put
|
|
|
|
def request(self, method, path, body=None, key=None):
|
|
self.calls.append((method, path))
|
|
if method == "GET" and path in ("api/apikeys", seq.KEY_LIST):
|
|
# Actual API may omit Token: probe must verify it independently.
|
|
visible = self.keys if path == seq.KEY_LIST else [v for v in self.keys if v.get("OwnerId") is not None]
|
|
return [{k: copy.deepcopy(v) for k, v in value.items() if k != "Token"} for value in visible]
|
|
if path == "api/apikeys/template":
|
|
return {"InputSettings": {"AppliedProperties": []}}
|
|
if path == "api/apikeys/":
|
|
self.assert_saved_before_post(body["Token"])
|
|
value = copy.deepcopy(body)
|
|
value["Id"] = "apikey-notification"
|
|
self.keys.append(value)
|
|
if self.fail_after_create:
|
|
raise RuntimeError("trusted Seq endpoint unavailable")
|
|
return copy.deepcopy(value)
|
|
if method == "PUT" and path.startswith("api/apikeys/"):
|
|
if self.fail_before_put:
|
|
raise RuntimeError("trusted Seq endpoint unavailable")
|
|
index = next(i for i, value in enumerate(self.keys) if path == "api/apikeys/" + value["Id"])
|
|
original_token = self.keys[index].get("Token")
|
|
self.keys[index] = copy.deepcopy(body)
|
|
self.keys[index]["Token"] = original_token
|
|
if self.fail_after_put:
|
|
raise RuntimeError("trusted Seq endpoint unavailable")
|
|
return copy.deepcopy(self.keys[index])
|
|
if path == "api/events/raw":
|
|
origin = next((value for value in self.keys if value.get("Token") == key), None)
|
|
if origin is None:
|
|
raise RuntimeError("Seq request denied (HTTP 401)")
|
|
canary = json.loads(body)
|
|
if set(canary) != {"@t", "@mt", "Application", "ProvisioningCheckId"}:
|
|
raise AssertionError("unexpected canary scope")
|
|
properties = {k: v for k, v in canary.items() if not k.startswith("@")}
|
|
properties.update({p["Name"]: p["Value"] for p in origin["InputSettings"]["AppliedProperties"]})
|
|
self.events.append({"ApiKeyId": origin["Id"], "Properties": [{"Name": k, "Value": v} for k, v in properties.items()]})
|
|
return None
|
|
if method == "GET" and path.startswith("api/events?count=2&filter="):
|
|
query = seq.urllib.parse.unquote(path.split("filter=", 1)[1])
|
|
check_id = query.split("'", 2)[1]
|
|
return [event for event in self.events if any(p == {"Name": "ProvisioningCheckId", "Value": check_id} for p in event["Properties"])]
|
|
raise AssertionError("unexpected API operation")
|
|
|
|
def assert_saved_before_post(self, token):
|
|
if self.token_file is None or self.token_file.read_text().strip() != token:
|
|
raise AssertionError("token was not persisted before create")
|
|
|
|
|
|
class SeqKeyTests(unittest.TestCase):
|
|
def test_create_preserves_token_before_post_and_does_not_touch_staging_key(self):
|
|
with tempfile.TemporaryDirectory() as folder:
|
|
path = Path(folder) / "private" / "seq-key.txt"
|
|
staging = {"Id": "apikey-staging", "Title": "invemory-staging-ingest", "AssignedPermissions": ["Ingest"]}
|
|
client = FakeSeq([staging], path)
|
|
self.assertEqual(seq.provision(client, path), "apikey-notification")
|
|
self.assertEqual(client.keys[0], staging)
|
|
self.assertEqual(client.keys[1]["AssignedPermissions"], ["Ingest"])
|
|
self.assertEqual(client.keys[1]["InputSettings"]["AppliedProperties"], seq.PROPERTIES + [{"Name": seq.MARKER, "Value": "apikey-notification"}])
|
|
self.assertEqual(path.read_text().strip(), client.keys[1]["Token"])
|
|
|
|
def test_repeat_verifies_existing_key_without_rotation(self):
|
|
with tempfile.TemporaryDirectory() as folder:
|
|
path = Path(folder) / "private" / "seq-key.txt"
|
|
client = FakeSeq(token_file=path)
|
|
seq.provision(client, path)
|
|
original = path.read_bytes()
|
|
client.calls.clear()
|
|
seq.provision(client, path)
|
|
self.assertEqual(path.read_bytes(), original)
|
|
self.assertEqual(len(client.keys), 1)
|
|
self.assertNotIn(("POST", "api/apikeys/"), client.calls)
|
|
self.assertFalse(any(method == "PUT" for method, _ in client.calls))
|
|
self.assertTrue(all(seq.allowed_path(method, path) for method, path in client.calls))
|
|
|
|
def test_ambiguous_keys_fail_before_token_generation(self):
|
|
with tempfile.TemporaryDirectory() as folder:
|
|
path = Path(folder) / "private" / "seq-key.txt"
|
|
with self.assertRaisesRegex(RuntimeError, "ambiguous"):
|
|
seq.provision(FakeSeq([key(), key()]), path)
|
|
self.assertFalse(path.exists())
|
|
|
|
def test_existing_key_with_extra_permissions_is_not_modified(self):
|
|
with tempfile.TemporaryDirectory() as folder:
|
|
path = Path(folder) / "private" / "seq-key.txt"
|
|
existing = key()
|
|
existing["AssignedPermissions"].append("Read")
|
|
client = FakeSeq([existing])
|
|
with self.assertRaisesRegex(RuntimeError, "only Ingest"):
|
|
seq.provision(client, path)
|
|
self.assertFalse(path.exists())
|
|
self.assertEqual(client.keys, [existing])
|
|
|
|
def test_existing_key_with_wrong_stamp_is_not_modified(self):
|
|
with tempfile.TemporaryDirectory() as folder:
|
|
path = Path(folder) / "private" / "seq-key.txt"
|
|
existing = key()
|
|
existing["InputSettings"]["AppliedProperties"][0]["Value"] = "Staging"
|
|
client = FakeSeq([existing])
|
|
with self.assertRaisesRegex(RuntimeError, "properties mismatch"):
|
|
seq.provision(client, path)
|
|
self.assertFalse(path.exists())
|
|
self.assertEqual(client.keys, [existing])
|
|
|
|
def test_missing_local_token_does_not_rotate_remote_key(self):
|
|
with tempfile.TemporaryDirectory() as folder:
|
|
path = Path(folder) / "private" / "seq-key.txt"
|
|
client = FakeSeq([key()])
|
|
with self.assertRaisesRegex(RuntimeError, "local token is missing"):
|
|
seq.provision(client, path)
|
|
self.assertFalse(path.exists())
|
|
self.assertEqual(client.calls, [("GET", seq.KEY_LIST)])
|
|
|
|
def test_unknown_post_outcome_is_recovered_with_original_token(self):
|
|
with tempfile.TemporaryDirectory() as folder:
|
|
path = Path(folder) / "private" / "seq-key.txt"
|
|
client = FakeSeq(token_file=path, fail_after_create=True)
|
|
with self.assertRaises(RuntimeError):
|
|
seq.provision(client, path)
|
|
original = path.read_bytes()
|
|
client.fail_after_create = False
|
|
seq.provision(client, path)
|
|
self.assertEqual(path.read_bytes(), original)
|
|
self.assertEqual(len(client.keys), 1)
|
|
|
|
def test_wrong_preserved_token_fails_probe_without_rotation(self):
|
|
with tempfile.TemporaryDirectory() as folder:
|
|
path = Path(folder) / "private" / "seq-key.txt"
|
|
seq.private_directory(path.parent)
|
|
path.write_text("b" * 64)
|
|
path.chmod(0o600)
|
|
client = FakeSeq([key()])
|
|
with self.assertRaisesRegex(RuntimeError, "HTTP 401"):
|
|
seq.provision(client, path)
|
|
self.assertEqual(path.read_text(), "b" * 64)
|
|
self.assertNotIn(("POST", "api/apikeys/"), client.calls)
|
|
|
|
def test_valid_foreign_staging_token_is_rejected_by_canary_identity(self):
|
|
with tempfile.TemporaryDirectory() as folder:
|
|
path = Path(folder) / "private" / "seq-key.txt"
|
|
seq.private_directory(path.parent)
|
|
path.write_text("b" * 64)
|
|
path.chmod(0o600)
|
|
staging = key("b" * 64)
|
|
staging.update(Id="apikey-staging", Title="invemory-staging-ingest")
|
|
staging["InputSettings"]["AppliedProperties"] = [{"Name": "Environment", "Value": "Staging"}]
|
|
client = FakeSeq([key(), staging])
|
|
with self.assertRaisesRegex(RuntimeError, "different or unverified key"):
|
|
seq.provision(client, path)
|
|
self.assertIn(("POST", "api/events/raw"), client.calls) # Valid ingestion alone is insufficient.
|
|
self.assertNotIn(("POST", "api/apikeys/"), client.calls)
|
|
self.assertEqual(path.read_text(), "b" * 64)
|
|
self.assertEqual(client.keys[1], staging)
|
|
self.assertEqual(client.keys[0]["Token"], key()["Token"])
|
|
|
|
def test_foreign_key_with_matching_properties_is_still_rejected(self):
|
|
with tempfile.TemporaryDirectory() as folder:
|
|
path = Path(folder) / "private" / "seq-key.txt"
|
|
seq.private_directory(path.parent)
|
|
path.write_text("b" * 64)
|
|
path.chmod(0o600)
|
|
foreign = key("b" * 64)
|
|
foreign.update(Id="apikey-foreign", Title="Other Service")
|
|
with self.assertRaisesRegex(RuntimeError, "different or unverified key"):
|
|
seq.provision(FakeSeq([key(), foreign]), path)
|
|
|
|
def test_server_stamped_marker_proves_identity_without_event_key_id_or_token(self):
|
|
client = FakeSeq()
|
|
check_id = "c" * 32
|
|
client.events = [{"Properties": [{"Name": "ProvisioningCheckId", "Value": check_id},
|
|
{"Name": seq.MARKER, "Value": "apikey-notification"}] + copy.deepcopy(seq.PROPERTIES)}]
|
|
redacted = key()
|
|
redacted["Token"] = None
|
|
seq.verify_canary(client, check_id, redacted)
|
|
|
|
def test_missing_server_stamped_marker_fails_closed(self):
|
|
client = FakeSeq()
|
|
check_id = "c" * 32
|
|
client.events = [{"Properties": [{"Name": "ProvisioningCheckId", "Value": check_id}] + copy.deepcopy(seq.PROPERTIES)}]
|
|
redacted = key()
|
|
redacted.pop("Token")
|
|
with self.assertRaisesRegex(RuntimeError, "unverified key"):
|
|
seq.verify_canary(client, check_id, redacted)
|
|
|
|
def test_correct_key_id_still_requires_actual_shared_stamps(self):
|
|
client = FakeSeq()
|
|
check_id = "c" * 32
|
|
client.events = [{"ApiKeyId": "apikey-notification", "Properties": [
|
|
{"Name": "ProvisioningCheckId", "Value": check_id}, {"Name": seq.MARKER, "Value": "apikey-notification"},
|
|
{"Name": "Environment", "Value": "Staging"}]}]
|
|
with self.assertRaisesRegex(RuntimeError, "stamped identity mismatch"):
|
|
seq.verify_canary(client, check_id, key())
|
|
|
|
def test_marker_collision_fails_before_put_or_ingestion(self):
|
|
with tempfile.TemporaryDirectory() as folder:
|
|
path = Path(folder) / "private" / "seq-key.txt"
|
|
seq.private_directory(path.parent)
|
|
path.write_text("a" * 64)
|
|
path.chmod(0o600)
|
|
foreign = key("b" * 64)
|
|
foreign.update(Id="apikey-foreign", Title="Other Service")
|
|
foreign["InputSettings"]["AppliedProperties"].append({"Name": seq.MARKER, "Value": "apikey-notification"})
|
|
client = FakeSeq([key(), foreign])
|
|
with self.assertRaisesRegex(RuntimeError, "marker collision"):
|
|
seq.provision(client, path)
|
|
self.assertFalse(any(method == "PUT" for method, _ in client.calls))
|
|
self.assertNotIn(("POST", "api/events/raw"), client.calls)
|
|
self.assertEqual(client.keys, [key(), foreign])
|
|
|
|
def test_shared_foreign_marker_collision_is_visible_in_full_listing(self):
|
|
with tempfile.TemporaryDirectory() as folder:
|
|
path = Path(folder) / "private" / "seq-key.txt"
|
|
seq.private_directory(path.parent)
|
|
path.write_text("a" * 64)
|
|
path.chmod(0o600)
|
|
dedicated = key()
|
|
dedicated["OwnerId"] = "user-operator"
|
|
shared = key("b" * 64)
|
|
shared.update(Id="apikey-shared-foreign", Title="Shared foreign service", OwnerId=None)
|
|
shared["InputSettings"]["AppliedProperties"].append({"Name": seq.MARKER, "Value": dedicated["Id"]})
|
|
client = FakeSeq([dedicated, shared])
|
|
# Demonstrate the narrower default endpoint would conceal the collision.
|
|
self.assertEqual(len(client.request("GET", "api/apikeys")), 1)
|
|
client.calls.clear()
|
|
with self.assertRaisesRegex(RuntimeError, "marker collision"):
|
|
seq.provision(client, path)
|
|
self.assertTrue(all(call == ("GET", seq.KEY_LIST) for call in client.calls))
|
|
self.assertEqual(client.keys, [dedicated, shared])
|
|
self.assertEqual(path.read_text(), "a" * 64)
|
|
|
|
def test_allowlist_requires_full_shared_key_listing(self):
|
|
self.assertTrue(seq.allowed_path("GET", "api/apikeys?shared=true"))
|
|
self.assertFalse(seq.allowed_path("GET", "api/apikeys"))
|
|
self.assertFalse(seq.allowed_path("GET", "api/apikeys?shared=false"))
|
|
|
|
def test_guarded_put_preserves_token_permissions_and_all_other_settings(self):
|
|
with tempfile.TemporaryDirectory() as folder:
|
|
path = Path(folder) / "private" / "seq-key.txt"
|
|
seq.private_directory(path.parent)
|
|
path.write_text("a" * 64)
|
|
path.chmod(0o600)
|
|
original = key()
|
|
original.update(OwnerId=None, IsDefault=False, TokenPrefix="aaaa")
|
|
original["InputSettings"].update(MinimumLevel="Information", Filter=None, CustomSettings={"retained": True})
|
|
expected = copy.deepcopy(original)
|
|
expected["InputSettings"]["AppliedProperties"].append({"Name": seq.MARKER, "Value": "apikey-notification"})
|
|
client = FakeSeq([original])
|
|
seq.provision(client, path)
|
|
self.assertEqual(client.keys, [expected])
|
|
self.assertEqual(path.read_text(), "a" * 64)
|
|
self.assertEqual([call for call in client.calls if call[0] == "PUT"], [("PUT", "api/apikeys/apikey-notification")])
|
|
|
|
def test_put_failure_preserves_original_key_and_local_token(self):
|
|
with tempfile.TemporaryDirectory() as folder:
|
|
path = Path(folder) / "private" / "seq-key.txt"
|
|
seq.private_directory(path.parent)
|
|
path.write_text("a" * 64)
|
|
path.chmod(0o600)
|
|
client = FakeSeq([key()], fail_before_put=True)
|
|
with self.assertRaisesRegex(RuntimeError, "unavailable"):
|
|
seq.provision(client, path)
|
|
self.assertEqual(client.keys, [key()])
|
|
self.assertEqual(path.read_text(), "a" * 64)
|
|
self.assertNotIn(("POST", "api/events/raw"), client.calls)
|
|
|
|
def test_unknown_put_outcome_is_recovered_without_another_put_or_rotation(self):
|
|
with tempfile.TemporaryDirectory() as folder:
|
|
path = Path(folder) / "private" / "seq-key.txt"
|
|
seq.private_directory(path.parent)
|
|
path.write_text("a" * 64)
|
|
path.chmod(0o600)
|
|
client = FakeSeq([key()], fail_after_put=True)
|
|
with self.assertRaisesRegex(RuntimeError, "unavailable"):
|
|
seq.provision(client, path)
|
|
client.calls.clear()
|
|
client.fail_after_put = False
|
|
seq.provision(client, path)
|
|
self.assertFalse(any(method == "PUT" for method, _ in client.calls))
|
|
self.assertNotIn(("POST", "api/apikeys/"), client.calls)
|
|
self.assertEqual(path.read_text(), "a" * 64)
|
|
self.assertEqual(len(client.keys), 1)
|
|
|
|
def test_redirect_is_rejected_before_credentials_can_be_forwarded(self):
|
|
with self.assertRaisesRegex(RuntimeError, "redirect denied"):
|
|
seq.NoRedirects().redirect_request(None, None, 302, "Found", {}, "https://untrusted.example/")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|