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()