Gecko [ v99win.today ]

Name	Size	Permission
_vendor	[ DIR ]	drwxr-xr-x
access_governance_cp	[ DIR ]	drwxr-xr-x
adm	[ DIR ]	drwxr-xr-x
ai_anomaly_detection	[ DIR ]	drwxr-xr-x
ai_document	[ DIR ]	drwxr-xr-x
ai_language	[ DIR ]	drwxr-xr-x
ai_speech	[ DIR ]	drwxr-xr-x
ai_vision	[ DIR ]	drwxr-xr-x
analytics	[ DIR ]	drwxr-xr-x
announcements_service	[ DIR ]	drwxr-xr-x
apigateway	[ DIR ]	drwxr-xr-x
apm_config	[ DIR ]	drwxr-xr-x
apm_control_plane	[ DIR ]	drwxr-xr-x
apm_synthetics	[ DIR ]	drwxr-xr-x
apm_traces	[ DIR ]	drwxr-xr-x
appmgmt_control	[ DIR ]	drwxr-xr-x
artifacts	[ DIR ]	drwxr-xr-x
audit	[ DIR ]	drwxr-xr-x
auth	[ DIR ]	drwxr-xr-x
autoscaling	[ DIR ]	drwxr-xr-x
bastion	[ DIR ]	drwxr-xr-x
bds	[ DIR ]	drwxr-xr-x
blockchain	[ DIR ]	drwxr-xr-x
budget	[ DIR ]	drwxr-xr-x
capacity_management	[ DIR ]	drwxr-xr-x
certificates	[ DIR ]	drwxr-xr-x
certificates_management	[ DIR ]	drwxr-xr-x
cims	[ DIR ]	drwxr-xr-x
circuit_breaker	[ DIR ]	drwxr-xr-x
cloud_bridge	[ DIR ]	drwxr-xr-x
cloud_guard	[ DIR ]	drwxr-xr-x
cloud_migrations	[ DIR ]	drwxr-xr-x
cluster_placement_groups	[ DIR ]	drwxr-xr-x
compute_cloud_at_customer	[ DIR ]	drwxr-xr-x
compute_instance_agent	[ DIR ]	drwxr-xr-x
container_engine	[ DIR ]	drwxr-xr-x
container_instances	[ DIR ]	drwxr-xr-x
core	[ DIR ]	drwxr-xr-x
dashboard_service	[ DIR ]	drwxr-xr-x
data_catalog	[ DIR ]	drwxr-xr-x
data_flow	[ DIR ]	drwxr-xr-x
data_integration	[ DIR ]	drwxr-xr-x
data_labeling_service	[ DIR ]	drwxr-xr-x
data_labeling_service_dataplan...	[ DIR ]	drwxr-xr-x
data_safe	[ DIR ]	drwxr-xr-x
data_science	[ DIR ]	drwxr-xr-x
database	[ DIR ]	drwxr-xr-x
database_management	[ DIR ]	drwxr-xr-x
database_migration	[ DIR ]	drwxr-xr-x
database_tools	[ DIR ]	drwxr-xr-x
delegate_access_control	[ DIR ]	drwxr-xr-x
demand_signal	[ DIR ]	drwxr-xr-x
desktops	[ DIR ]	drwxr-xr-x
devops	[ DIR ]	drwxr-xr-x
disaster_recovery	[ DIR ]	drwxr-xr-x
dns	[ DIR ]	drwxr-xr-x
dts	[ DIR ]	drwxr-xr-x
em_warehouse	[ DIR ]	drwxr-xr-x
email	[ DIR ]	drwxr-xr-x
email_data_plane	[ DIR ]	drwxr-xr-x
encryption	[ DIR ]	drwxr-xr-x
events	[ DIR ]	drwxr-xr-x
file_storage	[ DIR ]	drwxr-xr-x
fleet_apps_management	[ DIR ]	drwxr-xr-x
fleet_software_update	[ DIR ]	drwxr-xr-x
functions	[ DIR ]	drwxr-xr-x
fusion_apps	[ DIR ]	drwxr-xr-x
generative_ai	[ DIR ]	drwxr-xr-x
generative_ai_agent	[ DIR ]	drwxr-xr-x
generative_ai_agent_runtime	[ DIR ]	drwxr-xr-x
generative_ai_inference	[ DIR ]	drwxr-xr-x
generic_artifacts_content	[ DIR ]	drwxr-xr-x
globally_distributed_database	[ DIR ]	drwxr-xr-x
golden_gate	[ DIR ]	drwxr-xr-x
governance_rules_control_plane	[ DIR ]	drwxr-xr-x
healthchecks	[ DIR ]	drwxr-xr-x
identity	[ DIR ]	drwxr-xr-x
identity_data_plane	[ DIR ]	drwxr-xr-x
identity_domains	[ DIR ]	drwxr-xr-x
integration	[ DIR ]	drwxr-xr-x
jms	[ DIR ]	drwxr-xr-x
jms_java_downloads	[ DIR ]	drwxr-xr-x
key_management	[ DIR ]	drwxr-xr-x
license_manager	[ DIR ]	drwxr-xr-x
limits	[ DIR ]	drwxr-xr-x
load_balancer	[ DIR ]	drwxr-xr-x
lockbox	[ DIR ]	drwxr-xr-x
log_analytics	[ DIR ]	drwxr-xr-x
logging	[ DIR ]	drwxr-xr-x
loggingingestion	[ DIR ]	drwxr-xr-x
loggingsearch	[ DIR ]	drwxr-xr-x
management_agent	[ DIR ]	drwxr-xr-x
management_dashboard	[ DIR ]	drwxr-xr-x
marketplace	[ DIR ]	drwxr-xr-x
marketplace_private_offer	[ DIR ]	drwxr-xr-x
marketplace_publisher	[ DIR ]	drwxr-xr-x
media_services	[ DIR ]	drwxr-xr-x
monitoring	[ DIR ]	drwxr-xr-x
mysql	[ DIR ]	drwxr-xr-x
network_firewall	[ DIR ]	drwxr-xr-x
network_load_balancer	[ DIR ]	drwxr-xr-x
nosql	[ DIR ]	drwxr-xr-x
object_storage	[ DIR ]	drwxr-xr-x
oce	[ DIR ]	drwxr-xr-x
oci_control_center	[ DIR ]	drwxr-xr-x
ocvp	[ DIR ]	drwxr-xr-x
oda	[ DIR ]	drwxr-xr-x
onesubscription	[ DIR ]	drwxr-xr-x
ons	[ DIR ]	drwxr-xr-x
opa	[ DIR ]	drwxr-xr-x
opensearch	[ DIR ]	drwxr-xr-x
operator_access_control	[ DIR ]	drwxr-xr-x
opsi	[ DIR ]	drwxr-xr-x
optimizer	[ DIR ]	drwxr-xr-x
os_management	[ DIR ]	drwxr-xr-x
os_management_hub	[ DIR ]	drwxr-xr-x
osp_gateway	[ DIR ]	drwxr-xr-x
osub_billing_schedule	[ DIR ]	drwxr-xr-x
osub_organization_subscription	[ DIR ]	drwxr-xr-x
osub_subscription	[ DIR ]	drwxr-xr-x
osub_usage	[ DIR ]	drwxr-xr-x
pagination	[ DIR ]	drwxr-xr-x
psql	[ DIR ]	drwxr-xr-x
queue	[ DIR ]	drwxr-xr-x
recovery	[ DIR ]	drwxr-xr-x
redis	[ DIR ]	drwxr-xr-x
resource_manager	[ DIR ]	drwxr-xr-x
resource_scheduler	[ DIR ]	drwxr-xr-x
resource_search	[ DIR ]	drwxr-xr-x
retry	[ DIR ]	drwxr-xr-x
rover	[ DIR ]	drwxr-xr-x
sch	[ DIR ]	drwxr-xr-x
secrets	[ DIR ]	drwxr-xr-x
security_attribute	[ DIR ]	drwxr-xr-x
service_catalog	[ DIR ]	drwxr-xr-x
service_manager_proxy	[ DIR ]	drwxr-xr-x
service_mesh	[ DIR ]	drwxr-xr-x
stack_monitoring	[ DIR ]	drwxr-xr-x
streaming	[ DIR ]	drwxr-xr-x
tenant_manager_control_plane	[ DIR ]	drwxr-xr-x
threat_intelligence	[ DIR ]	drwxr-xr-x
usage	[ DIR ]	drwxr-xr-x
usage_api	[ DIR ]	drwxr-xr-x
vault	[ DIR ]	drwxr-xr-x
vbs_inst	[ DIR ]	drwxr-xr-x
visual_builder	[ DIR ]	drwxr-xr-x
vn_monitoring	[ DIR ]	drwxr-xr-x
vulnerability_scanning	[ DIR ]	drwxr-xr-x
waa	[ DIR ]	drwxr-xr-x
waas	[ DIR ]	drwxr-xr-x
waf	[ DIR ]	drwxr-xr-x
work_requests	[ DIR ]	drwxr-xr-x
zpr	[ DIR ]	drwxr-xr-x
__init__.py	6.14 KB	-rw-r--r--
alloy.py	6.52 KB	-rw-r--r--
base_client.py	50.76 KB	-rw-r--r--
config.py	9.98 KB	-rw-r--r--
constants.py	487 B	-rw-r--r--
decorators.py	1.61 KB	-rw-r--r--
exceptions.py	8.67 KB	-rw-r--r--
fips.py	3.17 KB	-rw-r--r--
regions.py	21.44 KB	-rw-r--r--
regions_definitions.py	6.45 KB	-rw-r--r--
request.py	1.97 KB	-rw-r--r--
response.py	2.19 KB	-rw-r--r--
service_endpoints.py	1.97 KB	-rw-r--r--
signer.py	11.49 KB	-rw-r--r--
util.py	13.09 KB	-rw-r--r--
version.py	367 B	-rw-r--r--
waiter.py	7.68 KB	-rw-r--r--

Code Editor : signer.py

# coding: utf-8
# Copyright (c) 2016, 2025, 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 __future__ import absolute_import

import base64
import email.utils
import hashlib
import io
import functools
import os

from oci._vendor import six
from oci.util import record_body_position_for_rewind, rewind_body, back_up_body_calculate_stream_content_length, read_stream_for_signing

from ._vendor import httpsig_cffi, requests
from .exceptions import InvalidPrivateKey, MissingPrivateKeyPassphrase

from cryptography.exceptions import UnsupportedAlgorithm
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives import serialization
import logging

logger = logging.getLogger(__name__)
SIGNATURE_VERSION = "1"

def load_private_key_from_file(filename, pass_phrase=None):
    filename = os.path.expanduser(filename)
    with io.open(filename, mode="rb") as f:
        private_key_data = f.read().strip()
    return load_private_key(private_key_data, pass_phrase)

def load_private_key(secret, pass_phrase):
    """Loads a private key that may use a pass_phrase.

Tries to correct or diagnose common errors:

- provided pass_phrase but didn't need one
    - provided a public key
    """
    if isinstance(secret, six.text_type):
        secret = secret.encode("ascii")
    if isinstance(pass_phrase, six.text_type):
        pass_phrase = pass_phrase.encode("ascii")

backend = default_backend()

try:
        # 0) Try with pass_phrase
        return serialization.load_pem_private_key(secret, pass_phrase, backend=backend)
    except TypeError:
        # 1) Either:
        #    - key has pass_phrase and one wasn't provided
        #    - key doesn't have pass_phrase and one was provided.
        #
        #    Can't fix the first, but we *can* fix the second.
        #    This can happen if the DEFAULT profile has a pass_phrase but
        #    another profile uses a key file without a pass_phrase.
        if pass_phrase is None:
            # 1.1) private key needed a pass_phrase and we don't have one
            raise MissingPrivateKeyPassphrase("The provided key requires a passphrase.")
        else:
            # 1.2) try again without pass_phrase; could be an artifact from DEFAULT
            return serialization.load_pem_private_key(secret, None, backend=backend)
    except ValueError:
        # 2) Try to determine what kind of failure this is.
        #    Most likely, this is either a bad password or a public key.
        #    If loading it as a public key fails, it's almost certainly a bad password.
        for loader in [
            serialization.load_der_public_key,
            serialization.load_pem_public_key,
            serialization.load_ssh_public_key
        ]:
            try:
                loader(secret, backend=backend)
            except (ValueError, UnsupportedAlgorithm):
                # 2.1) Not a public key; try the next format
                pass
            else:
                # 2.2) This is a public key
                raise InvalidPrivateKey("Authentication requires a private key, but a public key was provided.")
        # 2.3) Password is probably wrong.
        raise InvalidPrivateKey("The provided key is not a private key, or the provided passphrase is incorrect.")

def inject_missing_headers(request, sign_body, enforce_content_headers):
    # Inject date, host, and content-type if missing
    request.headers.setdefault(
        "date", email.utils.formatdate(usegmt=True))

request.headers.setdefault(
        "host", six.moves.urllib.parse.urlparse(request.url).netloc)

if hasattr(request.body, "buffer") or hasattr(request.body, "read"):
        request.headers.setdefault("content-type", "application/octet-stream")
    request.headers.setdefault("content-type", "application/json")

if enforce_content_headers:
        # Requests with a body need to send content-type,
        # content-length, and x-content-sha256
        if "x-content-sha256" not in request.headers and sign_body:
            body = request.body or ""
            m = hashlib.sha256()
            # Handle String types
            if isinstance(body, six.string_types):
                body = body.encode("utf-8")
                request.headers.setdefault("content-length", str(len(body)))
                m.update(body)
            # Handle bytes
            elif isinstance(body, (bytes, bytearray)):
                m.update(body)
            # Handling signing for Files/stdin
            elif hasattr(body, "buffer") or hasattr(body, "read"):
                is_body_rewindable, original_position = record_body_position_for_rewind(body)
                if is_body_rewindable:
                    content_length = read_stream_for_signing(m, body)
                    if content_length == -1:
                        raise IOError("Unable to read stream for signing! Please sign the stream yourself by using the custom header x-content-sha256")
                    request.headers.setdefault("content-length", str(content_length))
                    is_rewind_success = rewind_body(body, original_position)
                    if not is_rewind_success:
                        raise IOError("Unable to rewind request body while signing!")
                else:
                    logger.warning("Stream cannot be rewound, trying to backup and sign the body!")
                    stream = back_up_body_calculate_stream_content_length(body)
                    # Updating request body as it cannot be rewound
                    request.body = stream.get("byte_content")
                    m.update(stream.get("byte_content"))
                    request.headers.setdefault("content-length", str(stream.get("content_length")))
            # Update sha256 header
            if m:
                base64digest = base64.b64encode(m.digest())
                base64string = base64digest.decode("utf-8")
                request.headers["x-content-sha256"] = base64string

# HeaderSigner doesn't support private keys with passwords.
# Patched since the constructor parses the key in __init__
class _PatchedHeaderSigner(httpsig_cffi.sign.HeaderSigner):
    HEADER_SIGNER_TEMPLATE = 'Signature algorithm="rsa-sha256",headers="{}",keyId="{}",signature="%s",version="{}"'

"""Internal.  If you need to construct a Signer, use :class:`~.Signer` instead."""
    def __init__(self, key_id, private_key, headers):
        # Dropped general support for the specific signing/hash the SDK uses.
        self.sign_algorithm = "rsa"
        self.hash_algorithm = "sha256"

self._hash = None
        self._rsahash = httpsig_cffi.utils.HASHES[self.hash_algorithm]

self._rsa_private = private_key
        self._rsa_public = self._rsa_private.public_key()

self.headers = headers
        self.signature_template = self.HEADER_SIGNER_TEMPLATE.format(" ".join(headers), key_id, SIGNATURE_VERSION)

def reset_signer(self, key_id, private_key):
        self._hash = None
        self._rsa_private = private_key
        self._rsa_public = self._rsa_private.public_key()
        self.signature_template = self.HEADER_SIGNER_TEMPLATE.format(" ".join(self.headers), key_id, SIGNATURE_VERSION)

# An abstract class whose subclasses can sign requests. This contains the core logic for creating a signer and signing
# requests, but does not source the required information:
#
#   - api key
#   - private key
#   - headers
#
# As concrete implementations are expected to provide these and have their ways of sourcing/constructing them.
class AbstractBaseSigner(requests.auth.AuthBase):
    def create_signers(self, api_key, private_key, generic_headers, body_headers):
        self._basic_signer = _PatchedHeaderSigner(
            key_id=api_key,
            private_key=private_key,
            headers=generic_headers)

self._body_signer = _PatchedHeaderSigner(
            key_id=api_key,
            private_key=private_key,
            headers=generic_headers + body_headers)

def validate_request(self, request):
        verb = request.method.lower()
        if verb not in ["get", "head", "delete", "put", "post", "patch"]:
            raise ValueError("Don't know how to sign request verb {}".format(verb))

def do_request_sign(self, request, enforce_content_headers=True):
        verb = request.method.lower()
        sign_body = verb in ["put", "post", "patch"]
        if sign_body and enforce_content_headers:
            signer = self._body_signer
        else:
            signer = self._basic_signer
            # The requests library sets the Transfer-Encoding header to 'chunked' if the
            # body is a stream with 0 length. Object storage does not currently support this option,
            # and the request will fail if it is not removed. This is the only hook available where we
            # can do this after the header is added and before the request is sent.
            request.headers.pop('Transfer-Encoding', None)

inject_missing_headers(request, sign_body, enforce_content_headers)
        signed_headers = signer.sign(
            request.headers,
            host=six.moves.urllib.parse.urlparse(request.url).netloc,
            method=request.method,
            path=request.path_url)
        request.headers.update(signed_headers)

return request

def __call__(self, request, enforce_content_headers=True):
        self.validate_request(request)
        return self.do_request_sign(request, enforce_content_headers)

@property
    def without_content_headers(self):
        return functools.partial(self, enforce_content_headers=False)

class Signer(AbstractBaseSigner):
    """
    A requests auth instance that can be reused across requests. This signer is intended to be used
    when signing requests for a given user and it requires that user's ID, their private key
    and cerificate fingerprint.

The private key can be sourced from a file (private_key_file_location) or the PEM string can be
    provided directly (private_key_content).

The headers to be signed by this signer are not customizable.

You can manually sign calls by creating an instance of the signer, and
    providing it as the ``auth`` argument to Requests functions:

.. code-block:: python

import requests
        from oci import Signer

auth = Signer(...)
        resp = requests.get("https://...", auth=auth)

"""

def __init__(self, tenancy, user, fingerprint, private_key_file_location, pass_phrase=None, private_key_content=None):
        self.api_key = tenancy + "/" + user + "/" + fingerprint

if private_key_content:
            self.private_key = load_private_key(private_key_content, pass_phrase)
        else:
            self.private_key = load_private_key_from_file(private_key_file_location, pass_phrase)

generic_headers = ["date", "(request-target)", "host"]
        body_headers = ["content-length", "content-type", "x-content-sha256"]
        self.create_signers(self.api_key, self.private_key, generic_headers, body_headers)

@staticmethod
    def from_config(config):
        from .config import validate_config
        validate_config(config)
        return Signer(
            config['tenancy'],
            config['user'],
            config['fingerprint'],
            private_key_file_location=config['key_file'],
            pass_phrase=config.get('pass_phrase'),
            private_key_content=config.get('key_content')
        )

${this.title}

Code Editor : signer.py