Coverage for src / apcore_cli / security / auth.py: 89%

1"""API key authentication (FE-05).""" 

2 

3from __future__ import annotations 

4 

5from typing import TYPE_CHECKING 

6 

7if TYPE_CHECKING: 

8 from apcore_cli.config import ConfigResolver 

9 from apcore_cli.security.config_encryptor import ConfigEncryptor 

10 

11 

12class AuthenticationError(Exception): 

13 pass 

14 

15 

16class AuthProvider: 

17 """Resolve and authenticate API keys for remote registry calls. 

18 

19 Audit D1-006 parity (v0.6.x): the optional `encryptor` parameter mirrors 

20 the TypeScript `AuthProvider(config, encryptor?)` constructor. When 

21 omitted, falls back to `config.encryptor` (if present) or constructs a 

22 `ConfigEncryptor()` lazily. 

23 """ 

24 

25 def __init__( 

26 self, 

27 config: ConfigResolver, 

28 encryptor: ConfigEncryptor | None = None, 

29 ) -> None: 

30 self._config = config 

31 self._encryptor = encryptor 

32 

33 def _get_encryptor(self) -> ConfigEncryptor: 

34 if self._encryptor is not None: 

35 return self._encryptor 

36 existing = getattr(self._config, "encryptor", None) 

37 if existing is not None: 

38 return existing # type: ignore[no-any-return] 

39 # Lazy construction — defer the import to avoid a hard cycle. 

40 from apcore_cli.security.config_encryptor import ConfigEncryptor as _ConfigEncryptor 

41 

42 return _ConfigEncryptor() 

43 

44 def get_api_key(self) -> str | None: 

45 result = self._config.resolve("auth.api_key", cli_flag="--api-key", env_var="APCORE_AUTH_API_KEY") 

46 if result is not None and (result.startswith("keyring:") or result.startswith("enc:")): 

47 from apcore_cli.security.config_encryptor import ConfigDecryptionError 

48 

49 try: 

50 result = self._get_encryptor().retrieve(result, "auth.api_key") 

51 except ConfigDecryptionError as exc: 

52 raise AuthenticationError( 

53 "Failed to decrypt stored API key. Re-configure with 'apcore-cli config set auth.api_key'." 

54 ) from exc 

55 return result 

56 

57 def authenticate_request(self, headers: dict) -> dict: 

58 key = self.get_api_key() 

59 if key is None: 

60 raise AuthenticationError( 

61 "Remote registry requires authentication. " 

62 "Set --api-key, APCORE_AUTH_API_KEY, or auth.api_key in config." 

63 ) 

64 if "\r" in key or "\n" in key: 

65 raise AuthenticationError( 

66 "Malformed API key: contains invalid characters (CR/LF). " 

67 "Re-configure with 'apcore-cli config set auth.api_key'." 

68 ) 

69 headers["Authorization"] = f"Bearer {key.strip()}" 

70 return headers 

71 

72 def handle_response(self, status_code: int) -> None: 

73 if status_code in (401, 403): 

74 raise AuthenticationError("Authentication failed. Verify your API key.")