Coverage for oracle / oci_fsdr_mcp_server / auth.py: 100.00%

1""" 

2Copyright (c) 2025, Oracle and/or its affiliates. 

3Licensed under the Universal Permissive License v1.0 as shown at 

4https://oss.oracle.com/licenses/upl. 

5 

6OCI client factory supporting API key and session token authentication. 

7 

8Authentication is controlled by the OCI_AUTH_TYPE environment variable: 

9 - "api_key" (default): standard OCI config file credentials 

10 - "security_token": OCI CLI session token (for interactive/cloud-shell sessions) 

11""" 

12 

13from __future__ import annotations 

14 

15import os 

16from typing import Dict 

17 

18import oci 

19 

20from . import __project__, __version__ 

21from .consts import DEFAULT_OCI_AUTH_TYPE, DEFAULT_OCI_CONFIG_FILE 

22 

23_user_agent_name = __project__.split("oracle.", 1)[1].split("-server", 1)[0] 

24_ADDITIONAL_UA = f"{_user_agent_name}/{__version__}" 

25 

26_client_cache: Dict[str, oci.disaster_recovery.DisasterRecoveryClient] = {} 

27 

28 

29def get_dr_client(profile: str) -> oci.disaster_recovery.DisasterRecoveryClient: 

30 """Return a cached DisasterRecoveryClient for the given OCI config profile. 

31 

32 Created lazily on first use to avoid any stdout writes before the MCP 

33 stdio transport is active. 

34 """ 

35 if profile not in _client_cache: 

36 _client_cache[profile] = _make_client(profile) 

37 return _client_cache[profile] 

38 

39 

40def _make_client(profile: str) -> oci.disaster_recovery.DisasterRecoveryClient: 

41 if DEFAULT_OCI_AUTH_TYPE == "security_token": 

42 return _make_security_token_client(profile) 

43 return _make_api_key_client(profile) 

44 

45 

46def _make_api_key_client(profile: str) -> oci.disaster_recovery.DisasterRecoveryClient: 

47 """Standard OCI config file / API key authentication.""" 

48 config = oci.config.from_file( 

49 file_location=DEFAULT_OCI_CONFIG_FILE, profile_name=profile 

50 ) 

51 oci.config.validate_config(config) 

52 config["additional_user_agent"] = _ADDITIONAL_UA 

53 return oci.disaster_recovery.DisasterRecoveryClient(config) 

54 

55 

56def _make_security_token_client(profile: str) -> oci.disaster_recovery.DisasterRecoveryClient: 

57 """OCI CLI session token authentication (security_token auth type). 

58 

59 Reads the token file path and private key from the OCI config profile. 

60 Falls back to environment variables OCI_SECURITY_TOKEN_FILE and OCI_KEY_FILE. 

61 """ 

62 config = oci.config.from_file( 

63 file_location=DEFAULT_OCI_CONFIG_FILE, profile_name=profile 

64 ) 

65 

66 token_file = os.path.expanduser( 

67 config.get("security_token_file") 

68 or os.getenv("OCI_SECURITY_TOKEN_FILE", "~/.oci/token") 

69 ) 

70 key_file = os.path.expanduser( 

71 config.get("key_file") 

72 or os.getenv("OCI_KEY_FILE", "~/.oci/oci_api_key.pem") 

73 ) 

74 

75 with open(token_file, encoding="utf-8") as f: 

76 token = f.read().strip() 

77 

78 signer = oci.auth.signers.SecurityTokenSigner( 

79 token=token, 

80 private_key_file_location=key_file, 

81 ) 

82 return oci.disaster_recovery.DisasterRecoveryClient( 

83 config={"additional_user_agent": _ADDITIONAL_UA}, signer=signer 

84 )