Aurora
Adminer
Auto Root
WP Admin
cPanel Reset
Anti Backdoor
Root
lib
python3.9
site-packages
oci
Upload
New Folder
New File
Name
Size
Permissions
Actions
..
-
-
-
Upload File
Select File
New Folder
Folder Name
New File
File Name
Add WordPress Admin
Database Host
Database Name
Database User
Database Password
Admin Username
Admin Password
cPanel Password Reset
Email Address
Edit: signer.py
# coding: utf-8 # Copyright (c) 2016, 2024, 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') )