Key OIDC identity by issuer and subject

server/tests/test_identity.py

@@ -0,0 +1,111 @@
+"""OIDC issuer and subject identity isolation tests."""
+from __future__ import annotations
+
+import unittest
+
+from kca.models import ApiToken, User
+from kca.tokens import MAX_TOKENS_PER_USER
+from tests.helpers.api import api_client
+from tests.helpers.asynctest import async_test
+from tests.helpers.oidc import oidc_user
+
+
+class IdentityIsolationTests(unittest.TestCase):
+    @async_test
+    async def test_same_subject_at_different_issuers_is_isolated(self) -> None:
+        issuer_a = "https://issuer-a.example"
+        issuer_b = "https://issuer-b.example"
+
+        async with api_client() as client:
+            with oidc_user("shared-subject", "Issuer A User", iss=issuer_a):
+                response = await client.post("/api/vaults", json={"name": "A vault"})
+                assert response.status_code == 201, response.text
+                vault_a = response.json()["id"]
+                response = await client.post(
+                    f"/api/vaults/{vault_a}/keys",
+                    json={"name": "A private key", "key_type": "ec", "curve": "secp256r1"},
+                )
+                assert response.status_code == 201, response.text
+                key_a = response.json()["id"]
+                response = await client.post("/api/tokens", json={"name": "A token"})
+                assert response.status_code == 201, response.text
+                token_a = response.json()
+
+            with oidc_user("shared-subject", "Issuer B User", iss=issuer_b):
+                response = await client.get("/api/vaults")
+                assert response.status_code == 200
+                assert response.json()["results"] == []
+
+                response = await client.get(f"/api/vaults/{vault_a}")
+                assert response.status_code == 404
+                response = await client.post(
+                    f"/api/vaults/{vault_a}/keys/{key_a}/export",
+                    json={"format": "pem-private"},
+                )
+                assert response.status_code == 404
+
+                response = await client.get("/api/tokens")
+                assert response.status_code == 200
+                assert response.json()["results"] == []
+                response = await client.delete(f"/api/tokens/{token_a['id']}")
+                assert response.status_code == 404
+
+                response = await client.post("/api/vaults", json={"name": "B vault"})
+                assert response.status_code == 201, response.text
+                vault_b = response.json()["id"]
+                response = await client.post("/api/tokens", json={"name": "B token"})
+                assert response.status_code == 201, response.text
+                token_b = response.json()
+
+            users = await User.filter(sub="shared-subject")
+            assert len(users) == 2
+            assert {user.issuer for user in users} == {issuer_a, issuer_b}
+            assert {user.display_name for user in users} == {"Issuer A User", "Issuer B User"}
+
+            for token, expected_vault in ((token_a, vault_a), (token_b, vault_b)):
+                response = await client.get(
+                    "/api/vaults",
+                    headers={"authorization": f"Bearer {token['token']}"},
+                )
+                assert response.status_code == 200, response.text
+                assert [vault["id"] for vault in response.json()["results"]] == [expected_vault]
+
+    @async_test
+    async def test_token_limit_is_scoped_to_issuer(self) -> None:
+        issuer_a = "https://issuer-a.example"
+        issuer_b = "https://issuer-b.example"
+
+        async with api_client() as client:
+            with oidc_user("limited-subject", iss=issuer_a):
+                response = await client.get("/api/me")
+                assert response.status_code == 200
+                await ApiToken.bulk_create(
+                    [
+                        ApiToken(
+                            user_issuer=issuer_a,
+                            user_sub="limited-subject",
+                            name=f"token {number}",
+                            token_prefix=f"prefix-{number}",
+                            token_hash=f"{number:064x}",
+                        )
+                        for number in range(MAX_TOKENS_PER_USER)
+                    ]
+                )
+                response = await client.post("/api/tokens", json={"name": "over limit"})
+                assert response.status_code == 400
+
+            with oidc_user("limited-subject", iss=issuer_b):
+                response = await client.post("/api/tokens", json={"name": "independent"})
+                assert response.status_code == 201, response.text
+
+    @async_test
+    async def test_identity_without_issuer_is_rejected(self) -> None:
+        async with api_client() as client:
+            with oidc_user("missing-issuer", iss=None):
+                response = await client.get("/api/me")
+                assert response.status_code == 401
+            assert await User.filter(sub="missing-issuer").count() == 0
+
+
+if __name__ == "__main__":
+    unittest.main()