Coverage for arclith / adapters / output / vault / secret_adapter.py: 90%

1from __future__ import annotations 

2 

3import os 

4from pathlib import Path 

5 

6from arclith.domain.ports.secret_resolver import SecretResolver 

7 

8 

9def _read_vault_token() -> str | None: 

10 token = os.environ.get("VAULT_TOKEN") 

11 if token: 

12 return token 

13 token_file = Path.home() / ".vault-token" 

14 if token_file.exists(): 14 ↛ 16line 14 didn't jump to line 16 because the condition on line 14 was always true

15 return token_file.read_text().strip() or None 

16 return None 

17 

18 

19class VaultSecretAdapter(SecretResolver): 

20 """Reads secrets from HashiCorp Vault KV v2. 

21 

22 Each secret_key is a Vault path relative to the mount point. 

23 The secret must expose its value under the field "value". 

24 

25 Example: 

26 vault kv put kv/rekipe/recipe/mongodb value="mongodb://..." 

27 """ 

28 

29 def __init__(self, addr: str, mount: str = "kv") -> None: 

30 self._addr = addr 

31 self._mount = mount 

32 

33 def get(self, field_path: str, secret_key: str) -> str | None: 

34 try: 

35 import hvac # type: ignore[import-untyped] 

36 except ImportError: 

37 return None 

38 

39 token = _read_vault_token() 

40 if not token: 40 ↛ 41line 40 didn't jump to line 41 because the condition on line 40 was never true

41 return None 

42 

43 try: 

44 client = hvac.Client(url=self._addr, token=token) 

45 if not client.is_authenticated(): 

46 return None 

47 response = client.secrets.kv.v2.read_secret_version( 

48 path=secret_key, 

49 mount_point=self._mount, 

50 raise_on_deleted_version=True, 

51 ) 

52 return response["data"]["data"].get("value") 

53 except Exception: 

54 return None 

55