tests/core/test_api_key_store.py¶
Source from this local checkout, regenerated when the reader rebuilds.
Line links use #L<number>; a GitHub line range opens its first line.
1 # Copyright 2025 Softwell S.r.l.2 #3 # Licensed under the Apache License, Version 2.0 (the "License");4 # you may not use this file except in compliance with the License.5 # You may obtain a copy of the License at6 #7 # https://www.apache.org/licenses/LICENSE-2.08 #9 # Unless required by applicable law or agreed to in writing, software10 # distributed under the License is distributed on an "AS IS" BASIS,11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.12 # See the License for the specific language governing permissions and13 # limitations under the License.14 15 """API key store tests (core 1b Phase 5): contract suite + file backend specifics.16 17 The contract suite is PARAMETRIZED over FACTORIES (invariant §5.9): callables18 returning a fresh configured store over the SAME mount, so a future db backend19 plugs into the SAME suite. Today the only backend is ``FileApiKeyStore`` over20 a tmp ``site`` mount with key material installed (records are ciphertext at rest).21 """22 23 from __future__ import annotations24 25 import json26 import time27 28 import pytest29 from cryptography.fernet import Fernet30 31 from genro_storage.exceptions import StorageError32 33 from tests.storage_support import site_storage34 35 from genro_asgi import ApiKeyStore, FileApiKeyStore36 from genro_asgi.auth.api_key_store import API_KEY_PREFIX37 38 # --- store contract suite (parametrized over FACTORIES, §5.9) ---39 40 41 def _file_factory(tmp_path):42 """A factory building fresh ``FileApiKeyStore``s over one shared encrypted tmp mount."""43 key = Fernet.generate_key().decode()44 45 def make(**kwargs):46 storage = site_storage(tmp_path)47 storage.set_encryption_keys(key)48 return FileApiKeyStore(storage, **kwargs)49 50 return make51 52 53 STORE_FACTORIES = [_file_factory]54 55 56 @pytest.fixture(params=STORE_FACTORIES)57 def store_factory(request, tmp_path):58 """A callable returning a fresh configured api key store."""59 return request.param(tmp_path)60 61 62 def _key_id(key: str) -> str:63 """Extract the embedded ``key_id`` out of a ``gak_<key_id>_<secret>`` key."""64 return key.removeprefix(API_KEY_PREFIX).partition("_")[0]65 66 67 class TestApiKeyStoreContract:68 def test_is_an_api_key_store(self, store_factory) -> None:69 assert isinstance(store_factory(), ApiKeyStore)70 71 def test_load_all_empty_when_no_keys(self, store_factory) -> None:72 assert store_factory().load_all() == []73 74 def test_get_unknown_returns_none(self, store_factory) -> None:75 assert store_factory().get("nobody") is None76 77 def test_delete_absent_returns_false(self, store_factory) -> None:78 assert store_factory().delete("ghost") is False79 80 def test_issue_returns_a_prefixed_key(self, store_factory) -> None:81 assert store_factory().issue("ci-deploy", ["deploy"]).startswith(API_KEY_PREFIX)82 83 def test_issue_verify_roundtrip(self, store_factory) -> None:84 store = store_factory()85 key = store.issue("ci-deploy", ["deploy"])86 record = store.verify(key)87 assert record is not None88 assert record["label"] == "ci-deploy"89 assert record["tags"] == ["deploy"]90 91 def test_load_all_returns_every_issued_key(self, store_factory) -> None:92 store = store_factory()93 store.issue("one", [])94 store.issue("two", [])95 assert {r["label"] for r in store.load_all()} == {"one", "two"}96 97 def test_verify_wrong_secret_fails(self, store_factory) -> None:98 store = store_factory()99 key = store.issue("ci-deploy", [])100 assert store.verify(key + "x") is None101 102 def test_verify_unknown_key_id_returns_none(self, store_factory) -> None:103 assert store_factory().verify(f"{API_KEY_PREFIX}deadbeef_somesecret") is None104 105 def test_verify_malformed_key_returns_none(self, store_factory) -> None:106 assert store_factory().verify("not-a-key-at-all") is None107 108 def test_revoked_key_fails_instantly(self, store_factory) -> None:109 store = store_factory()110 key = store.issue("ci-deploy", [])111 assert store.revoke(_key_id(key)) is True112 assert store.verify(key) is None113 114 def test_revoke_absent_returns_false(self, store_factory) -> None:115 assert store_factory().revoke("ghost") is False116 117 def test_expired_key_fails(self, store_factory) -> None:118 store = store_factory()119 key = store.issue("ci-deploy", [], expires_at=time.time() - 10)120 assert store.verify(key) is None121 122 def test_unexpired_key_succeeds(self, store_factory) -> None:123 store = store_factory()124 key = store.issue("ci-deploy", [], expires_at=time.time() + 3600)125 assert store.verify(key) is not None126 127 def test_disabled_key_fails(self, store_factory) -> None:128 store = store_factory()129 key = store.issue("ci-deploy", [])130 record = store.get(_key_id(key))131 assert record is not None132 record["enabled"] = False133 store.save(record)134 assert store.verify(key) is None135 136 def test_delete_removes_record(self, store_factory) -> None:137 store = store_factory()138 key = store.issue("ci-deploy", [])139 key_id = _key_id(key)140 assert store.delete(key_id) is True141 assert store.get(key_id) is None142 143 144 # --- FileApiKeyStore specifics (persistence + ciphertext at rest) ---145 146 147 def _encrypted_storage(tmp_path, key):148 """The site storage over ``tmp_path`` with ``key`` installed as at-rest key material."""149 return site_storage(tmp_path, storage_key=key)150 151 152 class TestFileApiKeyStore:153 def test_record_survives_a_new_store_on_the_same_mount(self, tmp_path) -> None:154 key_material = Fernet.generate_key().decode()155 store = FileApiKeyStore(_encrypted_storage(tmp_path, key_material))156 key = store.issue("ci-deploy", ["deploy"])157 fresh = FileApiKeyStore(_encrypted_storage(tmp_path, key_material))158 record = fresh.verify(key)159 assert record is not None160 assert record["label"] == "ci-deploy"161 162 def test_on_disk_payload_is_ciphertext(self, tmp_path) -> None:163 key_material = Fernet.generate_key().decode()164 store = FileApiKeyStore(_encrypted_storage(tmp_path, key_material))165 key = store.issue("ci-deploy", ["deploy"])166 raw = (tmp_path / "api_keys" / f"{_key_id(key)}.json").read_bytes()167 assert raw.startswith(b"#GNRE1:") # the self-describing envelope168 assert b"ci-deploy" not in raw169 with pytest.raises(json.JSONDecodeError):170 json.loads(raw)171 assert Fernet(key_material).decrypt(raw.split(b"\n", 1)[1])172 173 def test_encrypted_write_without_keys_raises_on_issue(self, tmp_path) -> None:174 store = FileApiKeyStore(site_storage(tmp_path))175 with pytest.raises(StorageError):176 store.issue("ci-deploy", [])