OXIESEC PANEL

Name	Size	Modified	Perms
📁 ..	-	02/11/2025 08:19:48 AM	rwxr-xr-x
📄 __init__.py	808 bytes	12/09/2024 05:26:03 PM	rw-r--r--
📁 __pycache__	-	02/11/2025 08:19:49 AM	rwxr-xr-x
📄 mock_cloud_api.py	7.97 KB	12/09/2024 05:26:03 PM	rw-r--r--
📄 mock_logging_handler.py	1.28 KB	12/09/2024 05:26:03 PM	rw-r--r--
📄 rewrite_helper.py	2.8 KB	12/09/2024 05:26:03 PM	rw-r--r--
📄 signurl_signatures.py	5.7 KB	12/09/2024 05:26:03 PM	rw-r--r--
📄 test_Doption.py	9.72 KB	12/09/2024 05:26:03 PM	rw-r--r--
📄 test_acl.py	55.99 KB	12/09/2024 05:26:03 PM	rw-r--r--
📄 test_autoclass.py	6.85 KB	12/09/2024 05:26:03 PM	rw-r--r--
📄 test_boto_util.py	9.65 KB	12/09/2024 05:26:03 PM	rw-r--r--
📄 test_bucketconfig.py	4.98 KB	12/09/2024 05:26:03 PM	rw-r--r--
📄 test_bucketpolicyonly.py	3.78 KB	12/09/2024 05:26:03 PM	rw-r--r--
📄 test_cat.py	11.55 KB	12/09/2024 05:26:03 PM	rw-r--r--
📄 test_cloud_api_delegator.py	2 KB	12/09/2024 05:26:03 PM	rw-r--r--
📄 test_command.py	3.39 KB	12/09/2024 05:26:03 PM	rw-r--r--
📄 test_command_runner.py	20.66 KB	12/09/2024 05:26:03 PM	rw-r--r--
📄 test_compose.py	14.12 KB	12/09/2024 05:26:03 PM	rw-r--r--
📄 test_context_config.py	18.78 KB	12/09/2024 05:26:03 PM	rw-r--r--
📄 test_copy_helper_funcs.py	39.76 KB	12/09/2024 05:26:03 PM	rw-r--r--
📄 test_copy_objects_iterator.py	4.49 KB	12/09/2024 05:26:03 PM	rw-r--r--
📄 test_cors.py	12.45 KB	12/09/2024 05:26:03 PM	rw-r--r--
📄 test_cp.py	216 KB	12/09/2024 05:26:03 PM	rw-r--r--
📄 test_creds_config.py	8.64 KB	12/09/2024 05:26:03 PM	rw-r--r--
📄 test_daisy_chain_wrapper.py	14.13 KB	12/09/2024 05:26:03 PM	rw-r--r--
📁 test_data	-	12/09/2024 05:26:03 PM	rwxr-xr-x
📄 test_defacl.py	14.36 KB	12/09/2024 05:26:03 PM	rw-r--r--
📄 test_defstorageclass.py	5.43 KB	12/09/2024 05:26:03 PM	rw-r--r--
📄 test_du.py	10.61 KB	12/09/2024 05:26:03 PM	rw-r--r--
📄 test_encryption_helper.py	4.62 KB	12/09/2024 05:26:03 PM	rw-r--r--
📄 test_execution_util.py	3.88 KB	12/09/2024 05:26:03 PM	rw-r--r--
📄 test_file_part.py	3.38 KB	12/09/2024 05:26:03 PM	rw-r--r--
📄 test_gcs_json_api.py	2.9 KB	12/09/2024 05:26:03 PM	rw-r--r--
📄 test_gcs_json_credentials.py	9.85 KB	12/09/2024 05:26:03 PM	rw-r--r--
📄 test_gcs_json_media.py	7.44 KB	12/09/2024 05:26:03 PM	rw-r--r--
📄 test_gsutil.py	4.68 KB	12/09/2024 05:26:03 PM	rw-r--r--
📄 test_hash.py	9.58 KB	12/09/2024 05:26:03 PM	rw-r--r--
📄 test_hashing_helper.py	10.78 KB	12/09/2024 05:26:03 PM	rw-r--r--
📄 test_help.py	3.5 KB	12/09/2024 05:26:03 PM	rw-r--r--
📄 test_hmac.py	23.9 KB	12/09/2024 05:26:03 PM	rw-r--r--
📄 test_iam.py	90.67 KB	12/09/2024 05:26:03 PM	rw-r--r--
📄 test_kms.py	16.68 KB	12/09/2024 05:26:03 PM	rw-r--r--
📄 test_label.py	11.55 KB	12/09/2024 05:26:03 PM	rw-r--r--
📄 test_lifecycle.py	13.8 KB	12/09/2024 05:26:03 PM	rw-r--r--
📄 test_logging.py	3.5 KB	12/09/2024 05:26:03 PM	rw-r--r--
📄 test_ls.py	53.16 KB	12/09/2024 05:26:03 PM	rw-r--r--
📄 test_mb.py	19.64 KB	12/09/2024 05:26:03 PM	rw-r--r--
📄 test_metrics.py	51.65 KB	12/09/2024 05:26:03 PM	rw-r--r--
📄 test_mtls.py	2.01 KB	12/09/2024 05:26:03 PM	rw-r--r--
📄 test_mv.py	13.04 KB	12/09/2024 05:26:03 PM	rw-r--r--
📄 test_naming.py	63.11 KB	12/09/2024 05:26:03 PM	rw-r--r--
📄 test_notification.py	5.9 KB	12/09/2024 05:26:03 PM	rw-r--r--
📄 test_notification_pubsub.py	5.46 KB	12/09/2024 05:26:03 PM	rw-r--r--
📄 test_pap.py	5.91 KB	12/09/2024 05:26:03 PM	rw-r--r--
📄 test_parallel_cp.py	10.15 KB	12/09/2024 05:26:03 PM	rw-r--r--
📄 test_parallelism_framework.py	33.09 KB	12/09/2024 05:26:03 PM	rw-r--r--
📄 test_perfdiag.py	12.62 KB	12/09/2024 05:26:03 PM	rw-r--r--
📄 test_plurality_checkable_iterator.py	7.53 KB	12/09/2024 05:26:03 PM	rw-r--r--
📄 test_posix_util.py	2.03 KB	12/09/2024 05:26:03 PM	rw-r--r--
📄 test_psc.py	5.88 KB	12/09/2024 05:26:03 PM	rw-r--r--
📄 test_rb.py	2.93 KB	12/09/2024 05:26:03 PM	rw-r--r--
📄 test_requester_pays.py	11.7 KB	12/09/2024 05:26:03 PM	rw-r--r--
📄 test_resumable_streaming.py	12.36 KB	12/09/2024 05:26:03 PM	rw-r--r--
📄 test_retention.py	28.9 KB	12/09/2024 05:26:03 PM	rw-r--r--
📄 test_retention_util.py	5.49 KB	12/09/2024 05:26:03 PM	rw-r--r--
📄 test_rewrite.py	31.62 KB	12/09/2024 05:26:03 PM	rw-r--r--
📄 test_rm.py	33.72 KB	12/09/2024 05:26:03 PM	rw-r--r--
📄 test_rpo.py	10.22 KB	12/09/2024 05:26:03 PM	rw-r--r--
📄 test_rsync.py	149.33 KB	12/09/2024 05:26:03 PM	rw-r--r--
📄 test_rsync_funcs.py	3.36 KB	12/09/2024 05:26:03 PM	rw-r--r--
📄 test_seek_ahead_thread.py	8.79 KB	12/09/2024 05:26:03 PM	rw-r--r--
📄 test_setmeta.py	12.54 KB	12/09/2024 05:26:03 PM	rw-r--r--
📄 test_shim_util.py	64.19 KB	12/09/2024 05:26:03 PM	rw-r--r--
📄 test_signurl.py	24.69 KB	12/09/2024 05:26:03 PM	rw-r--r--
📄 test_stat.py	11.38 KB	12/09/2024 05:26:03 PM	rw-r--r--
📄 test_stet_cp.py	5.77 KB	12/09/2024 05:26:03 PM	rw-r--r--
📄 test_stet_util.py	7.38 KB	12/09/2024 05:26:03 PM	rw-r--r--
📄 test_storage_url.py	7.02 KB	12/09/2024 05:26:03 PM	rw-r--r--
📄 test_tabcomplete.py	14.31 KB	12/09/2024 05:26:03 PM	rw-r--r--
📄 test_temporary_file_util.py	1.54 KB	12/09/2024 05:26:03 PM	rw-r--r--
📄 test_trace.py	1.76 KB	12/09/2024 05:26:03 PM	rw-r--r--
📄 test_tracker_file.py	9.9 KB	12/09/2024 05:26:03 PM	rw-r--r--
📄 test_ubla.py	3.88 KB	12/09/2024 05:26:03 PM	rw-r--r--
📄 test_ui.py	67.42 KB	12/09/2024 05:26:03 PM	rw-r--r--
📄 test_update.py	10.37 KB	12/09/2024 05:26:03 PM	rw-r--r--
📄 test_user_agent_helper.py	5.34 KB	12/09/2024 05:26:03 PM	rw-r--r--
📄 test_util.py	19.85 KB	12/09/2024 05:26:03 PM	rw-r--r--
📄 test_versioning.py	3.61 KB	12/09/2024 05:26:03 PM	rw-r--r--
📄 test_web.py	6.54 KB	12/09/2024 05:26:03 PM	rw-r--r--
📄 test_wildcard_iterator.py	22.18 KB	12/09/2024 05:26:03 PM	rw-r--r--
📄 test_wrapped_credentials.py	13.95 KB	12/09/2024 05:26:03 PM	rw-r--r--
📁 testcase	-	12/09/2024 05:26:03 PM	rwxr-xr-x
📄 util.py	29.01 KB	12/09/2024 05:26:03 PM	rw-r--r--

Editing: test_wrapped_credentials.py

# -*- coding: utf-8 -*-
# Copyright 2022 Google LLC. 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.
"""Tests for wrapped_credentials.py."""

import datetime
import json
import httplib2

from google.auth import aws
from google.auth import external_account
from google.auth import external_account_authorized_user
from google.auth import identity_pool
from google.auth import pluggable
from gslib.tests import testcase
from gslib.utils.wrapped_credentials import WrappedCredentials
import oauth2client

from six import add_move, MovedModule

add_move(MovedModule("mock", "mock", "unittest.mock"))
from six.moves import mock

ACCESS_TOKEN = "foo"
CONTENT = "content"
RESPONSE = httplib2.Response({
    "content-type": "text/plain",
    "status": "200",
    "content-length": len(CONTENT),
})

class MockCredentials(external_account.Credentials):

def __init__(self, token=None, expiry=None, *args, **kwargs):
    super().__init__(*args, **kwargs)
    self._audience = None
    self.expiry = expiry
    self.token = None

def side_effect(*args, **kwargs):
      del args, kwargs  # Unused.
      self.token = token

self.refresh = mock.Mock(side_effect=side_effect)

def retrieve_subject_token():
    pass

class HeadersWithAuth(dict):
  """A utility class to use to make sure a set of headers includes specific authentication"""

def __init__(self, token):
    self.token = token or ""

def __eq__(self, headers):
    return headers[b"Authorization"] == bytes("Bearer " + self.token, "utf-8")

class TestWrappedCredentials(testcase.GsUtilUnitTestCase):
  """Test logic for interacting with Wrapped Credentials the way we intend to use them."""

@mock.patch.object(httplib2, "Http", autospec=True)
  def testWrappedCredentialUsage(self, http):
    http.return_value.request.return_value = (RESPONSE, CONTENT)
    req = http.return_value.request

creds = WrappedCredentials(
        MockCredentials(token=ACCESS_TOKEN,
                        audience="foo",
                        subject_token_type="bar",
                        token_url="https://sts.googleapis.com",
                        credential_source="qux"))

http = oauth2client.transport.get_http_object()
    creds.authorize(http)
    _, content = http.request(uri="google.com")
    self.assertEqual(content, CONTENT)
    creds._base.refresh.assert_called_once_with(mock.ANY)

# Make sure the default request gets called with the correct token.
    req.assert_called_once_with("google.com",
                                method="GET",
                                headers=HeadersWithAuth(ACCESS_TOKEN),
                                body=None,
                                connection_type=mock.ANY,
                                redirections=mock.ANY)

def testWrappedCredentialSerialization(self):
    """Test logic for converting Wrapped Credentials to and from JSON for serialization."""
    creds = WrappedCredentials(
        identity_pool.Credentials(audience="foo",
                                  subject_token_type="bar",
                                  token_url="https://sts.googleapis.com",
                                  credential_source={"url": "google.com"}))
    creds.access_token = ACCESS_TOKEN
    creds.token_expiry = datetime.datetime(2001, 12, 5, 0, 0)
    creds_json = creds.to_json()
    json_values = json.loads(creds_json)
    self.assertEqual(json_values["client_id"], "foo")
    self.assertEqual(json_values['access_token'], ACCESS_TOKEN)
    self.assertEqual(json_values['token_expiry'], "2001-12-05T00:00:00Z")
    self.assertEqual(json_values["_base"]["audience"], "foo")
    self.assertEqual(json_values["_base"]["subject_token_type"], "bar")
    self.assertEqual(json_values["_base"]["token_url"],
                     "https://sts.googleapis.com")
    self.assertEqual(json_values["_base"]["credential_source"]["url"],
                     "google.com")

creds2 = WrappedCredentials.from_json(creds_json)
    self.assertIsInstance(creds2, WrappedCredentials)
    self.assertIsInstance(creds2._base, identity_pool.Credentials)
    self.assertEqual(creds2.client_id, "foo")
    self.assertEqual(creds2.access_token, ACCESS_TOKEN)
    self.assertEqual(creds2.token_expiry, creds.token_expiry)

def testWrappedCredentialSerializationMissingKeywords(self):
    """Test logic for creating a Wrapped Credentials using keywords that exist in IdentityPool but not AWS."""
    creds = WrappedCredentials.from_json(
        json.dumps({
            "client_id": "foo",
            "access_token": ACCESS_TOKEN,
            "token_expiry": "2001-12-05T00:00:00Z",
            "_base": {
                "type": "external_account",
                "audience": "foo",
                "subject_token_type": "bar",
                "token_url": "https://sts.googleapis.com",
                "credential_source": {
                    "url": "google.com",
                    "workforce_pool_user_project": "1234567890"
                }
            }
        }))

self.assertIsInstance(creds, WrappedCredentials)
    self.assertIsInstance(creds._base, identity_pool.Credentials)

@mock.patch.object(httplib2, "Http", autospec=True)
  def testWrappedCredentialUsageExternalAccountAuthorizedUser(self, http):
    http.return_value.request.return_value = (RESPONSE, CONTENT)
    req = http.return_value.request

creds = WrappedCredentials(
        external_account_authorized_user.Credentials(
            audience=
            "//iam.googleapis.com/locations/global/workforcePools/$WORKFORCE_POOL_ID/providers/$PROVIDER_ID",
            refresh_token="refreshToken",
            token_url="https://sts.googleapis.com/v1/oauth/token",
            token_info_url="https://sts.googleapis.com/v1/instrospect",
            client_id="clientId",
            client_secret="clientSecret"))

def _refresh_token_side_effect(*args, **kwargs):
      del args, kwargs  # Unused.
      creds._base.token = ACCESS_TOKEN

creds._base.refresh = mock.Mock(side_effect=_refresh_token_side_effect)

def testWrappedCredentialSerializationExternalAccountAuthorizedUser(self):
    """Test logic for converting Wrapped Credentials to and from JSON for serialization."""
    creds = WrappedCredentials(
        external_account_authorized_user.Credentials(
            audience=
            "//iam.googleapis.com/locations/global/workforcePools/$WORKFORCE_POOL_ID/providers/$PROVIDER_ID",
            refresh_token="refreshToken",
            token_url="https://sts.googleapis.com/v1/oauth/token",
            token_info_url="https://sts.googleapis.com/v1/instrospect",
            client_id="clientId",
            client_secret="clientSecret"))
    creds.access_token = ACCESS_TOKEN
    creds.token_expiry = datetime.datetime(2001, 12, 5, 0, 0)
    creds_json = creds.to_json()
    json_values = json.loads(creds_json)
    expected_json_values = {
        "_class": "WrappedCredentials",
        "_module": "gslib.utils.wrapped_credentials",
        "client_id": "clientId",
        "access_token": ACCESS_TOKEN,
        "token_expiry": "2001-12-05T00:00:00Z",
        "client_secret": "clientSecret",
        "refresh_token": "refreshToken",
        "id_token": None,
        "id_token_jwt": None,
        "invalid": False,
        "revoke_uri": None,
        "scopes": [],
        "token_info_uri": None,
        "token_response": None,
        "token_uri": None,
        "user_agent": None,
        "_base": {
            "type":
                "external_account_authorized_user",
            "audience":
                "//iam.googleapis.com/locations/global/workforcePools/$WORKFORCE_POOL_ID/providers/$PROVIDER_ID",
            "token":
                ACCESS_TOKEN,
            "expiry":
                "2001-12-05T00:00:00Z",
            "token_url":
                "https://sts.googleapis.com/v1/oauth/token",
            "token_info_url":
                "https://sts.googleapis.com/v1/instrospect",
            "refresh_token":
                "refreshToken",
            "client_id":
                "clientId",
            "client_secret":
                "clientSecret",
        }
    }
    self.assertEqual(json_values, expected_json_values)

creds2 = WrappedCredentials.from_json(creds_json)
    self.assertIsInstance(creds2, WrappedCredentials)
    self.assertIsInstance(creds2._base,
                          external_account_authorized_user.Credentials)
    self.assertEqual(creds2.client_id, "clientId")

def testFromJsonAWSCredentials(self):
    creds = WrappedCredentials.from_json(
        json.dumps({
            "_base": {
                "audience":
                    "//iam.googleapis.com/projects/123456/locations/global/workloadIdentityPools/POOL_ID/providers/PROVIDER_ID",
                "credential_source": {
                    "environment_id":
                        "aws1",
                    "region_url":
                        "http://169.254.169.254/latest/meta-data/placement/availability-zone",
                    "regional_cred_verification_url":
                        "https://sts.{region}.amazonaws.com?Action=GetCallerIdentity&Version=2011-06-15",
                    "url":
                        "http://169.254.169.254/latest/meta-data/iam/security-credentials"
                },
                "service_account_impersonation_url":
                    "https://iamcredentials.googleapis.com/v1/projects/-/serviceAccounts/service-1234@service-name.iam.gserviceaccount.com:generateAccessToken",
                "subject_token_type":
                    "urn:ietf:params:aws:token-type:aws4_request",
                "token_url":
                    "https://sts.googleapis.com/v1/token",
                "type":
                    "external_account"
            }
        }))

self.assertIsInstance(creds, WrappedCredentials)
    self.assertIsInstance(creds._base, external_account.Credentials)
    self.assertIsInstance(creds._base, aws.Credentials)

def testFromJsonFileBasedCredentials(self):
    creds = WrappedCredentials.from_json(
        json.dumps({
            "_base": {
                "audience":
                    "//iam.googleapis.com/projects/123456/locations/global/workloadIdentityPools/POOL_ID/providers/PROVIDER_ID",
                "credential_source": {
                    "file": "/var/run/secrets/goog.id/token"
                },
                "service_account_impersonation_url":
                    "https://iamcredentials.googleapis.com/v1/projects/-/serviceAccounts/service-1234@service-name.iam.gserviceaccount.com:generateAccessToken",
                "subject_token_type":
                    "urn:ietf:params:oauth:token-type:jwt",
                "token_url":
                    "https://sts.googleapis.com/v1/token",
                "type":
                    "external_account"
            }
        }))

self.assertIsInstance(creds, WrappedCredentials)
    self.assertIsInstance(creds._base, external_account.Credentials)
    self.assertIsInstance(creds._base, identity_pool.Credentials)

def testFromJsonPluggableCredentials(self):
    creds = WrappedCredentials.from_json(
        json.dumps({
            "_base": {
                "audience":
                    "//iam.googleapis.com/projects/123456/locations/global/workloadIdentityPools/POOL_ID/providers/PROVIDER_ID",
                "credential_source": {
                    "executable": {
                        "command": "/path/to/command.sh"
                    }
                },
                "service_account_impersonation_url":
                    "https://iamcredentials.googleapis.com/v1/projects/-/serviceAccounts/service-1234@service-name.iam.gserviceaccount.com:generateAccessToken",
                "subject_token_type":
                    "urn:ietf:params:oauth:token-type:jwt",
                "token_url":
                    "https://sts.googleapis.com/v1/token",
                "type":
                    "external_account"
            }
        }))

self.assertIsInstance(creds, WrappedCredentials)
    self.assertIsInstance(creds._base, external_account.Credentials)
    self.assertIsInstance(creds._base, pluggable.Credentials)

def testFromJsonExternalAccountAuthorizedUserCredentials(self):
    creds = WrappedCredentials.from_json(
        json.dumps({
            "_base": {
                "type":
                    "external_account_authorized_user",
                "audience":
                    "//iam.googleapis.com/locations/global/workforcePools/$WORKFORCE_POOL_ID/providers/$PROVIDER_ID",
                "refresh_token":
                    "refreshToken",
                "token_url":
                    "https://sts.googleapis.com/v1/oauth/token",
                "token_info_url":
                    "https://sts.googleapis.com/v1/instrospect",
                "client_id":
                    "clientId",
                "client_secret":
                    "clientSecret",
            }
        }))

self.assertIsInstance(creds, WrappedCredentials)
    self.assertIsInstance(creds._base,
                          external_account_authorized_user.Credentials)