File manager

File manager - Edit - /usr/lib/mysqlsh/lib/python3.9/site-packages/oci/auth/signers/instance_principals_security_token_signer.py

Back
# coding: utf-8 # Copyright (c) 2016, 2026, Oracle and/or its affiliates. All rights reserved. # This software is dual-licensed to you under the Universal Permissive License (UPL) 1.0 as shown at https://oss.oracle.com/licenses/upl or Apache License 2.0 as shown at http://www.apache.org/licenses/LICENSE-2.0. You may choose either license. from .security_token_signer import X509FederationClientBasedSecurityTokenSigner from ..certificate_retriever import UrlBasedCertificateRetriever, INSTANCE_METADATA_URL_CERTIFICATE_RETRIEVER_RETRY_STRATEGY from ..session_key_supplier import SessionKeySupplier from ..federation_client import X509FederationClient from .. import auth_utils from oci._vendor import requests from oci._vendor.requests.exceptions import HTTPError from oci._vendor.requests.exceptions import ConnectTimeout import oci.regions import os import logging class InstancePrincipalsSecurityTokenSigner(X509FederationClientBasedSecurityTokenSigner): """ A SecurityTokenSigner which uses a security token for an instance principal. This signer can also refresh its token as needed. This signer is self-sufficient in that its internals know how to source the required information to request and use the token: * Using the metadata endpoint for the instance (http://169.254.169.254/opc/v2) we can discover the region the instance is in, its leaf certificate and any intermediate certificates (for requesting the token) and the tenancy (as) that is in the leaf certificate. * You can override the metadata endpoint by overriding the environment variable OCI_METADATA_BASE_URL with an endpoint of your choice * The signer leverages X509FederationClient so it can refresh the security token and also get the private key needed to sign requests (via the client's session_key_supplier) This signer can be used as follows: .. code-block:: python import oci signer = oci.auth.signers.InstancePrincipalsSecurityTokenSigner() identity_client = oci.identity.IdentityClient(config={}, signer=signer) regions = identity_client.list_regions() :param str federation_endpoint: (optional) Users of this class can specify an explicit federation_endpoint, representing the Auth Service endpoint from which to retrieve a security token. If it is not provided then we will construct an endpoint based on the region we read from the metadata endpoint for the instance :param federation_client_cert_bundle_verify: (optional) If we need a specific cert bundle in order to perform verification against the federation endpoint, this parameter is the path to that bundle. Alternatively, False can be passed to disable verification. :type federation_client_cert_bundle_verify: str or Boolean :param obj federation_client_retry_strategy: (optional): A retry strategy to apply to calls made by the X509FederationClient used by this class. This should be one of the strategies available in the :py:mod:`~oci.retry` module. A convenience :py:data:`~oci.retry.DEFAULT_FEDERATION_CLIENT_RETRY_STRATEGY` is also available and will be used by the X509FederationClient if no explicit retry strategy is specified. The specifics of the default retry strategy are described `here <https://docs.oracle.com/en-us/iaas/tools/python/latest/sdk_behaviors/retries.html>`__. To have this operation explicitly not perform any retries, pass an instance of :py:class:`~oci.retry.NoneRetryStrategy`. :param bool log_requests: (optional) log_request if set to True, will log the request url and response data when retrieving the certificate & corresponding private key (if there is one defined for this retriever) from UrlBasedCertificateRetriever, region, federation endpoint, and the response for receiving the token from the federation endpoint """ METADATA_URL_BASE_ENV_VAR = 'OCI_METADATA_BASE_URL' DEFAULT_METADATA_URL_BASE = 'http://169.254.169.254/opc/v2' METADATA_URL_BASE = os.environ.get(METADATA_URL_BASE_ENV_VAR, DEFAULT_METADATA_URL_BASE) GET_REGION_URL = '{}/instance/region'.format(METADATA_URL_BASE) LEAF_CERTIFICATE_URL = '{}/identity/cert.pem'.format(METADATA_URL_BASE) LEAF_CERTIFICATE_PRIVATE_KEY_URL = '{}/identity/key.pem'.format(METADATA_URL_BASE) INTERMEDIATE_CERTIFICATE_URL = '{}/identity/intermediate.pem'.format(METADATA_URL_BASE) METADATA_AUTH_HEADERS = {'Authorization': 'Bearer Oracle'} def __init__(self, **kwargs): self.logger = logging.getLogger("{}.{}".format(__name__, id(self))) self.logger.addHandler(logging.NullHandler()) if kwargs.get('log_requests'): self.logger.disabled = False self.logger.setLevel(logging.DEBUG) else: self.logger.disabled = True # If user sends in custom retry strategy use that, else use INSTANCE_METADATA_URL_CERTIFICATE_RETRIEVER_RETRY_STRATEGY if kwargs.get('retry_strategy'): self.retry_strategy = kwargs['retry_strategy'] else: self.retry_strategy = INSTANCE_METADATA_URL_CERTIFICATE_RETRIEVER_RETRY_STRATEGY try: self.leaf_certificate_retriever = UrlBasedCertificateRetriever( certificate_url=self.LEAF_CERTIFICATE_URL, private_key_url=self.LEAF_CERTIFICATE_PRIVATE_KEY_URL, retry_strategy=self.retry_strategy, headers=self.METADATA_AUTH_HEADERS, log_requests=kwargs.get('log_requests')) except (HTTPError, ConnectTimeout, oci.exceptions.ServiceError) as e: if not e.args: e.args = ('',) e.args = e.args + ("Instance principals authentication can only be used on OCI compute instances. Please confirm this code is running on an OCI compute instance. See https://docs.oracle.com/en-us/iaas/Content/Identity/Tasks/callingservicesfrominstances.htm for more info.",) raise e self.intermediate_certificate_retriever = UrlBasedCertificateRetriever( certificate_url=self.INTERMEDIATE_CERTIFICATE_URL, retry_strategy=self.retry_strategy, headers=self.METADATA_AUTH_HEADERS) self.session_key_supplier = SessionKeySupplier() self.tenancy_id = auth_utils.get_tenancy_id_from_certificate(self.leaf_certificate_retriever.get_certificate_as_certificate()) self.initialize_and_return_region() if 'federation_endpoint' in kwargs and kwargs['federation_endpoint']: federation_endpoint = kwargs['federation_endpoint'] else: federation_endpoint = '{}/v1/x509'.format(oci.regions.endpoint_for('auth', self.region)) self.logger.debug("federation endpoint is set to : %s " % (federation_endpoint)) federation_client = X509FederationClient( federation_endpoint=federation_endpoint, tenancy_id=self.tenancy_id, session_key_supplier=self.session_key_supplier, leaf_certificate_retriever=self.leaf_certificate_retriever, intermediate_certificate_retrievers=[self.intermediate_certificate_retriever], cert_bundle_verify=kwargs.get('federation_client_cert_bundle_verify', None), retry_strategy=kwargs.get('federation_client_retry_strategy', None), purpose=kwargs.get('purpose', None), log_requests=kwargs.get('log_requests') ) if 'generic_headers' in kwargs: generic_headers = kwargs['generic_headers'] super(InstancePrincipalsSecurityTokenSigner, self).__init__(federation_client=federation_client, generic_headers=generic_headers) else: super(InstancePrincipalsSecurityTokenSigner, self).__init__(federation_client) def initialize_and_return_region(self): if hasattr(self, 'region'): return self.region if self.retry_strategy: self.retry_strategy.make_retrying_call(self._set_region_from_imds) else: self._set_region_from_imds() self.logger.debug(f"Region is set to: {self.region}") return self.region def _set_region_from_imds(self): self.logger.debug(f"Requesting region information from IMDS: {self.GET_REGION_URL}") response = requests.get(self.GET_REGION_URL, timeout=(10, 60), headers=self.METADATA_AUTH_HEADERS) try: response.raise_for_status() except HTTPError as e: raise oci.exceptions.ServiceError(e.response.status_code, e.errno, e.response.headers, str(e)) region_raw = response.text.strip().lower() # The region can be something like "phx" but internally we expect "us-phoenix-1", "us-ashburn-1" etc. if region_raw in oci.regions.REGIONS_SHORT_NAMES: self.region = oci.regions.REGIONS_SHORT_NAMES[region_raw] else: self.region = region_raw

| ver. 1.4 | Github | . | PHP 8.2.32 | Generation time: 0.06 | proxy | phpinfo | Settings