.

. BSA HACKER
Server : Apache
System : Linux nusantara.hosteko.com 4.18.0-553.16.1.lve.el8.x86_64 #1 SMP Tue Aug 13 17:45:03 UTC 2024 x86_64
User : koperas1 ( 1254)
PHP Version : 7.4.33
Disable Function : NONE
Directory : /proc/thread-self/root/opt/cppython/lib/python3.8/site-packages/oauth2client/
Upload File :
Current File : //proc/thread-self/root/opt/cppython/lib/python3.8/site-packages/oauth2client/crypt.py
# -*- coding: utf-8 -*-
#
# Copyright 2014 Google Inc. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Crypto-related routines for oauth2client."""

import json
import logging
import time

from oauth2client import _helpers
from oauth2client import _pure_python_crypt


RsaSigner = _pure_python_crypt.RsaSigner
RsaVerifier = _pure_python_crypt.RsaVerifier

CLOCK_SKEW_SECS = 300  # 5 minutes in seconds
AUTH_TOKEN_LIFETIME_SECS = 300  # 5 minutes in seconds
MAX_TOKEN_LIFETIME_SECS = 86400  # 1 day in seconds

logger = logging.getLogger(__name__)


class AppIdentityError(Exception):
    """Error to indicate crypto failure."""


def _bad_pkcs12_key_as_pem(*args, **kwargs):
    raise NotImplementedError('pkcs12_key_as_pem requires OpenSSL.')


try:
    from oauth2client import _openssl_crypt
    OpenSSLSigner = _openssl_crypt.OpenSSLSigner
    OpenSSLVerifier = _openssl_crypt.OpenSSLVerifier
    pkcs12_key_as_pem = _openssl_crypt.pkcs12_key_as_pem
except ImportError:  # pragma: NO COVER
    OpenSSLVerifier = None
    OpenSSLSigner = None
    pkcs12_key_as_pem = _bad_pkcs12_key_as_pem

try:
    from oauth2client import _pycrypto_crypt
    PyCryptoSigner = _pycrypto_crypt.PyCryptoSigner
    PyCryptoVerifier = _pycrypto_crypt.PyCryptoVerifier
except ImportError:  # pragma: NO COVER
    PyCryptoVerifier = None
    PyCryptoSigner = None


if OpenSSLSigner:
    Signer = OpenSSLSigner
    Verifier = OpenSSLVerifier
elif PyCryptoSigner:  # pragma: NO COVER
    Signer = PyCryptoSigner
    Verifier = PyCryptoVerifier
else:  # pragma: NO COVER
    Signer = RsaSigner
    Verifier = RsaVerifier


def make_signed_jwt(signer, payload, key_id=None):
    """Make a signed JWT.

    See http://self-issued.info/docs/draft-jones-json-web-token.html.

    Args:
        signer: crypt.Signer, Cryptographic signer.
        payload: dict, Dictionary of data to convert to JSON and then sign.
        key_id: string, (Optional) Key ID header.

    Returns:
        string, The JWT for the payload.
    """
    header = {'typ': 'JWT', 'alg': 'RS256'}
    if key_id is not None:
        header['kid'] = key_id

    segments = [
        _helpers._urlsafe_b64encode(_helpers._json_encode(header)),
        _helpers._urlsafe_b64encode(_helpers._json_encode(payload)),
    ]
    signing_input = b'.'.join(segments)

    signature = signer.sign(signing_input)
    segments.append(_helpers._urlsafe_b64encode(signature))

    logger.debug(str(segments))

    return b'.'.join(segments)


def _verify_signature(message, signature, certs):
    """Verifies signed content using a list of certificates.

    Args:
        message: string or bytes, The message to verify.
        signature: string or bytes, The signature on the message.
        certs: iterable, certificates in PEM format.

    Raises:
        AppIdentityError: If none of the certificates can verify the message
                          against the signature.
    """
    for pem in certs:
        verifier = Verifier.from_string(pem, is_x509_cert=True)
        if verifier.verify(message, signature):
            return

    # If we have not returned, no certificate confirms the signature.
    raise AppIdentityError('Invalid token signature')


def _check_audience(payload_dict, audience):
    """Checks audience field from a JWT payload.

    Does nothing if the passed in ``audience`` is null.

    Args:
        payload_dict: dict, A dictionary containing a JWT payload.
        audience: string or NoneType, an audience to check for in
                  the JWT payload.

    Raises:
        AppIdentityError: If there is no ``'aud'`` field in the payload
                          dictionary but there is an ``audience`` to check.
        AppIdentityError: If the ``'aud'`` field in the payload dictionary
                          does not match the ``audience``.
    """
    if audience is None:
        return

    audience_in_payload = payload_dict.get('aud')
    if audience_in_payload is None:
        raise AppIdentityError(
            'No aud field in token: {0}'.format(payload_dict))
    if audience_in_payload != audience:
        raise AppIdentityError('Wrong recipient, {0} != {1}: {2}'.format(
            audience_in_payload, audience, payload_dict))


def _verify_time_range(payload_dict):
    """Verifies the issued at and expiration from a JWT payload.

    Makes sure the current time (in UTC) falls between the issued at and
    expiration for the JWT (with some skew allowed for via
    ``CLOCK_SKEW_SECS``).

    Args:
        payload_dict: dict, A dictionary containing a JWT payload.

    Raises:
        AppIdentityError: If there is no ``'iat'`` field in the payload
                          dictionary.
        AppIdentityError: If there is no ``'exp'`` field in the payload
                          dictionary.
        AppIdentityError: If the JWT expiration is too far in the future (i.e.
                          if the expiration would imply a token lifetime
                          longer than what is allowed.)
        AppIdentityError: If the token appears to have been issued in the
                          future (up to clock skew).
        AppIdentityError: If the token appears to have expired in the past
                          (up to clock skew).
    """
    # Get the current time to use throughout.
    now = int(time.time())

    # Make sure issued at and expiration are in the payload.
    issued_at = payload_dict.get('iat')
    if issued_at is None:
        raise AppIdentityError(
            'No iat field in token: {0}'.format(payload_dict))
    expiration = payload_dict.get('exp')
    if expiration is None:
        raise AppIdentityError(
            'No exp field in token: {0}'.format(payload_dict))

    # Make sure the expiration gives an acceptable token lifetime.
    if expiration >= now + MAX_TOKEN_LIFETIME_SECS:
        raise AppIdentityError(
            'exp field too far in future: {0}'.format(payload_dict))

    # Make sure (up to clock skew) that the token wasn't issued in the future.
    earliest = issued_at - CLOCK_SKEW_SECS
    if now < earliest:
        raise AppIdentityError('Token used too early, {0} < {1}: {2}'.format(
            now, earliest, payload_dict))
    # Make sure (up to clock skew) that the token isn't already expired.
    latest = expiration + CLOCK_SKEW_SECS
    if now > latest:
        raise AppIdentityError('Token used too late, {0} > {1}: {2}'.format(
            now, latest, payload_dict))


def verify_signed_jwt_with_certs(jwt, certs, audience=None):
    """Verify a JWT against public certs.

    See http://self-issued.info/docs/draft-jones-json-web-token.html.

    Args:
        jwt: string, A JWT.
        certs: dict, Dictionary where values of public keys in PEM format.
        audience: string, The audience, 'aud', that this JWT should contain. If
                  None then the JWT's 'aud' parameter is not verified.

    Returns:
        dict, The deserialized JSON payload in the JWT.

    Raises:
        AppIdentityError: if any checks are failed.
    """
    jwt = _helpers._to_bytes(jwt)

    if jwt.count(b'.') != 2:
        raise AppIdentityError(
            'Wrong number of segments in token: {0}'.format(jwt))

    header, payload, signature = jwt.split(b'.')
    message_to_sign = header + b'.' + payload
    signature = _helpers._urlsafe_b64decode(signature)

    # Parse token.
    payload_bytes = _helpers._urlsafe_b64decode(payload)
    try:
        payload_dict = json.loads(_helpers._from_bytes(payload_bytes))
    except:
        raise AppIdentityError('Can\'t parse token: {0}'.format(payload_bytes))

    # Verify that the signature matches the message.
    _verify_signature(message_to_sign, signature, certs.values())

    # Verify the issued at and created times in the payload.
    _verify_time_range(payload_dict)

    # Check audience.
    _check_audience(payload_dict, audience)

    return payload_dict