Coverage for src/bayernwerk_client/tokens.py: 100%

1"""Generic JWT-based token storage, shared across services. 

2 

3Both `map` and `efix` authenticate via a Salesforce-issued JWT with an 

4`exp` claim and cache it locally the same way, so this lives here rather 

5than under either service. Each service passes its own distinct 

6`TokenStore` path (e.g. `map-tokens.json` / `efix-tokens.json`) - two 

7services sharing the generic default path below would silently overwrite 

8each other's cached token. 

9""" 

10 

11from __future__ import annotations 

12 

13import json 

14import os 

15import time 

16from dataclasses import asdict, dataclass 

17from pathlib import Path 

18 

19from bayernwerk_client._jwt import decode_jwt_payload 

20 

21DEFAULT_TOKEN_PATH = Path.home() / ".cache" / "bayernwerk-client" / "tokens.json" 

22"""Generic fallback for ad-hoc use - services define and pass their own path.""" 

23 

24# Refresh a bit before actual expiry to avoid racing a request against expiry. 

25EXPIRY_LEEWAY_SECONDS = 60 

26 

27 

28@dataclass 

29class TokenSet: 

30 access_token: str 

31 refresh_token: str | None = None 

32 id_token: str | None = None 

33 expires_at: float | None = None 

34 """Unix timestamp, decoded from the access token's `exp` claim.""" 

35 

36 @classmethod 

37 def from_access_token( 

38 cls, access_token: str, *, refresh_token: str | None = None, id_token: str | None = None 

39 ) -> TokenSet: 

40 expires_at: float | None = None 

41 try: 

42 expires_at = float(decode_jwt_payload(access_token)["exp"]) 

43 except (ValueError, KeyError, TypeError): 

44 expires_at = None 

45 return cls(access_token=access_token, refresh_token=refresh_token, id_token=id_token, expires_at=expires_at) 

46 

47 @property 

48 def is_expired(self) -> bool: 

49 if self.expires_at is None: 

50 return False 

51 return time.time() >= (self.expires_at - EXPIRY_LEEWAY_SECONDS) 

52 

53 

54class TokenStore: 

55 """Reads/writes a `TokenSet` to a local JSON file with restrictive permissions.""" 

56 

57 def __init__(self, path: Path | str = DEFAULT_TOKEN_PATH) -> None: 

58 self.path = Path(path) 

59 

60 def load(self) -> TokenSet | None: 

61 if not self.path.exists(): 

62 return None 

63 data = json.loads(self.path.read_text()) 

64 return TokenSet(**data) 

65 

66 def save(self, tokens: TokenSet) -> None: 

67 self.path.parent.mkdir(parents=True, exist_ok=True) 

68 self.path.write_text(json.dumps(asdict(tokens))) 

69 os.chmod(self.path, 0o600) 

70 

71 def clear(self) -> None: 

72 self.path.unlink(missing_ok=True)