external.py 6.9 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165
  1. import asyncio
  2. import calendar
  3. import json
  4. import logging
  5. import os
  6. import shlex
  7. import subprocess
  8. import time
  9. from typing import Callable, Optional
  10. from alibabacloud_credentials.provider.refreshable import Credentials, RefreshResult, RefreshCachedSupplier
  11. from alibabacloud_credentials_api import ICredentialsProvider
  12. from alibabacloud_credentials.exceptions import CredentialException
  13. log = logging.getLogger('credentials')
  14. ExternalCredentialUpdateCallback = Callable[[str, str, str, int], None]
  15. ExternalCredentialUpdateCallbackAsync = Callable[[str, str, str, int], None]
  16. def _parse_expiration(expiration: str) -> int:
  17. if not expiration:
  18. return 0
  19. time_array = time.strptime(expiration, '%Y-%m-%dT%H:%M:%SZ')
  20. return calendar.timegm(time_array)
  21. def _get_stale_time(expiration: int) -> int:
  22. if expiration <= 0:
  23. return int(time.mktime(time.localtime()))
  24. return expiration - 180
  25. class ExternalCredentialsProvider(ICredentialsProvider):
  26. DEFAULT_TIMEOUT = 60
  27. def __init__(self, *,
  28. process_command: str = None,
  29. timeout: int = None,
  30. credential_update_callback: Optional[ExternalCredentialUpdateCallback] = None,
  31. credential_update_callback_async: Optional[ExternalCredentialUpdateCallbackAsync] = None):
  32. if not process_command:
  33. raise ValueError('process_command is empty')
  34. self._process_command = process_command
  35. self._timeout = timeout if timeout and timeout > 0 else ExternalCredentialsProvider.DEFAULT_TIMEOUT
  36. self._credential_update_callback = credential_update_callback
  37. self._credential_update_callback_async = credential_update_callback_async
  38. self._credentials_cache = RefreshCachedSupplier(
  39. refresh_callable=self._refresh_credentials,
  40. refresh_callable_async=self._refresh_credentials_async,
  41. )
  42. def get_credentials(self) -> Credentials:
  43. return self._credentials_cache._sync_call()
  44. async def get_credentials_async(self) -> Credentials:
  45. return await self._credentials_cache._async_call()
  46. def _refresh_credentials(self) -> RefreshResult[Credentials]:
  47. if not self._process_command.strip():
  48. raise CredentialException('process_command is empty')
  49. try:
  50. command = self._process_command if os.name == 'nt' else shlex.split(self._process_command)
  51. completed = subprocess.run(
  52. command,
  53. stdout=subprocess.PIPE,
  54. stderr=subprocess.PIPE,
  55. timeout=self._timeout,
  56. check=False,
  57. text=True,
  58. shell=os.name == 'nt',
  59. )
  60. except subprocess.TimeoutExpired:
  61. raise CredentialException(f'command process timed out after {self._timeout * 1000} milliseconds')
  62. except Exception as e:
  63. raise CredentialException(f'failed to execute external command: {e}')
  64. if completed.returncode != 0:
  65. raise CredentialException(
  66. f'failed to execute external command: exit status {completed.returncode}\nstderr: {completed.stderr}')
  67. return self._parse_and_build_credentials(completed.stdout, async_callback=False)
  68. async def _refresh_credentials_async(self) -> RefreshResult[Credentials]:
  69. if not self._process_command.strip():
  70. raise CredentialException('process_command is empty')
  71. try:
  72. if os.name == 'nt':
  73. process = await asyncio.create_subprocess_shell(
  74. self._process_command,
  75. stdout=asyncio.subprocess.PIPE,
  76. stderr=asyncio.subprocess.PIPE,
  77. )
  78. else:
  79. process = await asyncio.create_subprocess_exec(
  80. *shlex.split(self._process_command),
  81. stdout=asyncio.subprocess.PIPE,
  82. stderr=asyncio.subprocess.PIPE,
  83. )
  84. stdout, stderr = await asyncio.wait_for(process.communicate(), timeout=self._timeout)
  85. except asyncio.TimeoutError:
  86. if 'process' in locals():
  87. process.kill()
  88. await process.wait()
  89. raise CredentialException(f'command process timed out after {self._timeout * 1000} milliseconds')
  90. except Exception as e:
  91. raise CredentialException(f'failed to execute external command: {e}')
  92. if process.returncode != 0:
  93. raise CredentialException(
  94. f'failed to execute external command: exit status {process.returncode}\nstderr: {stderr.decode("utf-8")}')
  95. return await self._parse_and_build_credentials_async(stdout.decode('utf-8'))
  96. def _parse_and_build_credentials(self, output: str, async_callback: bool) -> RefreshResult[Credentials]:
  97. try:
  98. data = json.loads(output)
  99. except Exception as e:
  100. raise CredentialException(f'failed to parse external command output: {e}')
  101. access_key_id = data.get('access_key_id')
  102. access_key_secret = data.get('access_key_secret')
  103. security_token = data.get('sts_token')
  104. if not access_key_id or not access_key_secret:
  105. raise CredentialException('invalid credential response: access_key_id or access_key_secret is empty')
  106. if data.get('mode') == 'StsToken' and not security_token:
  107. raise CredentialException('invalid StsToken credential response: sts_token is empty')
  108. expiration = _parse_expiration(data.get('expiration'))
  109. credentials = Credentials(
  110. access_key_id=access_key_id,
  111. access_key_secret=access_key_secret,
  112. security_token=security_token,
  113. expiration=expiration,
  114. provider_name=self.get_provider_name(),
  115. )
  116. if not async_callback and self._credential_update_callback:
  117. try:
  118. self._credential_update_callback(access_key_id, access_key_secret, security_token, expiration)
  119. except Exception as e:
  120. log.warning(f'failed to update external credentials in config file: {e}')
  121. return RefreshResult(value=credentials, stale_time=_get_stale_time(expiration))
  122. async def _parse_and_build_credentials_async(self, output: str) -> RefreshResult[Credentials]:
  123. result = self._parse_and_build_credentials(output, async_callback=True)
  124. credentials = result.value()
  125. if self._credential_update_callback_async:
  126. try:
  127. await self._credential_update_callback_async(
  128. credentials.get_access_key_id(),
  129. credentials.get_access_key_secret(),
  130. credentials.get_security_token(),
  131. credentials.get_expiration() or 0,
  132. )
  133. except Exception as e:
  134. log.warning(f'failed to update external credentials in config file: {e}')
  135. return result
  136. def get_provider_name(self) -> str:
  137. return 'external'