feat(auth): implement in-place Regional Access Boundary configuration and add public RAB getters

packages/google-auth/google/auth/credentials.py

@@ -309,6 +309,16 @@ def __init__(self):
             _regional_access_boundary_utils._RegionalAccessBoundaryManager()
         )
 
+    @property
+    def regional_access_boundary(self):
+        """Optional[str]: The encoded Regional Access Boundary locations."""
+        return self._rab_manager._data.encoded_locations
+
+    @property
+    def regional_access_boundary_expiry(self):
+        """Optional[datetime.datetime]: The expiration time of the Regional Access Boundary."""
+        return self._rab_manager._data.expiry
+
     @abc.abstractmethod
     def _perform_refresh_token(self, request):
         """Refreshes the access token.
@@ -361,39 +371,37 @@ def _copy_regional_access_boundary_manager(self, target):
         new_manager._data = self._rab_manager._data
         target._rab_manager = new_manager
 
-    def _with_regional_access_boundary(self, seed):
-        """Returns a copy of these credentials with the the regional_access_boundary
-        set to the provided seed. This is intended for internal use only as invalid
+    def _set_regional_access_boundary(self, seed):
+        """Applies the regional_access_boundary provided via the seed on these
+        credentials. This is intended for internal use only as invalid
         seeds would produce unexpected results until automatic recovery is supported.
         Currently this is used by the gcloud CLI and therefore changes to the
         contract MUST be backwards compatible (e.g. the method signature must be
-        unchanged and a copy of the credenials with the RAB set must be returned).
+        unchanged and the credentials with the RAB set must be returned).
 
 
         Returns:
-            google.auth.credentials.Credentials: A new credentials instance.
+            google.auth.credentials.Credentials: The credentials instance.
         """
-        creds = self._make_copy()
-        creds._rab_manager.set_initial_regional_access_boundary(
+        self._rab_manager.set_initial_regional_access_boundary(
             encoded_locations=seed.get("encodedLocations", None),
             expiry=seed.get("expiry", None),
         )
-        return creds
+        return self
 
-    def _with_blocking_regional_access_boundary_lookup(self):
-        """Returns a copy of these credentials with the blocking lookup mode enabled.
+    def _set_blocking_regional_access_boundary_lookup(self):
+        """Enables the blocking lookup mode on these credentials.
         This is intended for internal use only as blocking lookup requires additional
         care and consideration. Currently this is used by the gcloud CLI and
         therefore changes to the contract MUST be backwards compatible (e.g. the
-        method signature must be unchanged and a copy of the credentials with the
+        method signature must be unchanged and the credentials with the
         blocking lookup flag set to true must be returned).
 
         Returns:
-            google.auth.credentials.Credentials: A new credentials instance.
+            google.auth.credentials.Credentials: The credentials instance.
         """
-        creds = self._make_copy()
-        creds._rab_manager.enable_blocking_lookup()
-        return creds
+        self._rab_manager.enable_blocking_lookup()
+        return self
 
     def _maybe_start_regional_access_boundary_refresh(self, request, url):
         """

packages/google-auth/tests/compute_engine/test_credentials.py

@@ -451,13 +451,13 @@ def test_refresh_with_agent_identity_opt_out_or_not_agent(
         kwargs = mock_metadata_get.call_args[1]
         assert "bindCertificateFingerprint" not in kwargs.get("params", {})
 
-    def test_with_blocking_regional_access_boundary_lookup(self):
+    def test_set_blocking_regional_access_boundary_lookup(self):
         creds = self.credentials
         assert not creds._rab_manager._use_blocking_regional_access_boundary_lookup
 
-        new_creds = creds._with_blocking_regional_access_boundary_lookup()
-        assert new_creds._rab_manager._use_blocking_regional_access_boundary_lookup
-        assert new_creds is not creds
+        new_creds = creds._set_blocking_regional_access_boundary_lookup()
+        assert creds._rab_manager._use_blocking_regional_access_boundary_lookup
+        assert new_creds is creds
 
 
 class TestIDTokenCredentials(object):

packages/google-auth/tests/oauth2/test_credentials.py

@@ -96,13 +96,13 @@ def test_default_state(self):
         assert credentials.rapt_token == self.RAPT_TOKEN
         assert credentials.refresh_handler is None
 
-    def test_with_blocking_regional_access_boundary_lookup(self):
+    def test_set_blocking_regional_access_boundary_lookup(self):
         creds = self.make_credentials()
         assert not creds._rab_manager._use_blocking_regional_access_boundary_lookup
 
-        new_creds = creds._with_blocking_regional_access_boundary_lookup()
-        assert new_creds._rab_manager._use_blocking_regional_access_boundary_lookup
-        assert new_creds is not creds
+        new_creds = creds._set_blocking_regional_access_boundary_lookup()
+        assert creds._rab_manager._use_blocking_regional_access_boundary_lookup
+        assert new_creds is creds
 
     def test_get_cred_info(self):
         credentials = self.make_credentials()

packages/google-auth/tests/test__regional_access_boundary_utils.py

@@ -224,23 +224,47 @@ def test_apply_headers_removes_header_if_empty(self):
         creds._rab_manager.apply_headers(headers)
         assert headers == {}
 
-    def test_with_blocking_regional_access_boundary_lookup(self):
+    def test_set_blocking_regional_access_boundary_lookup(self):
         creds = CredentialsImpl()
         assert not creds._rab_manager._use_blocking_regional_access_boundary_lookup
 
-        new_creds = creds._with_blocking_regional_access_boundary_lookup()
-        assert new_creds._rab_manager._use_blocking_regional_access_boundary_lookup
+        new_creds = creds._set_blocking_regional_access_boundary_lookup()
+        assert new_creds is creds
+        assert creds._rab_manager._use_blocking_regional_access_boundary_lookup
 
-    def test_with_regional_access_boundary(self):
+    def test_set_regional_access_boundary(self):
         creds = CredentialsImpl()
         seed = {
             "encodedLocations": "0xABC",
             "expiry": _helpers.utcnow() + datetime.timedelta(hours=1),
         }
-        new_creds = creds._with_regional_access_boundary(seed)
-        assert new_creds._rab_manager._data.encoded_locations == "0xABC"
-        assert new_creds._rab_manager._data.expiry == seed["expiry"]
-        assert new_creds._rab_manager._data.cooldown_expiry is None
+        new_creds = creds._set_regional_access_boundary(seed)
+        assert new_creds is creds
+        assert creds._rab_manager._data.encoded_locations == "0xABC"
+        assert creds._rab_manager._data.expiry == seed["expiry"]
+        assert creds._rab_manager._data.cooldown_expiry is None
+
+    def test_regional_access_boundary_getter(self):
+        creds = CredentialsImpl()
+        assert creds.regional_access_boundary is None
+
+        seed = {
+            "encodedLocations": "0xABC",
+            "expiry": _helpers.utcnow() + datetime.timedelta(hours=1),
+        }
+        creds._set_regional_access_boundary(seed)
+        assert creds.regional_access_boundary == "0xABC"
+
+    def test_regional_access_boundary_expiry_getter(self):
+        creds = CredentialsImpl()
+        assert creds.regional_access_boundary_expiry is None
+
+        seed = {
+            "encodedLocations": "0xABC",
+            "expiry": _helpers.utcnow() + datetime.timedelta(hours=1),
+        }
+        creds._set_regional_access_boundary(seed)
+        assert creds.regional_access_boundary_expiry == seed["expiry"]
 
     def test_copy_regional_access_boundary_state(self):
         source_creds = CredentialsImpl()

packages/google-auth/tests/test_credentials.py

@@ -403,7 +403,7 @@ def test_before_request_triggers_rab_refresh():
             lookup.return_value = {"encodedLocations": "0xA30"}
 
             creds = CredentialsImpl()
-            creds = creds._with_blocking_regional_access_boundary_lookup()
+            creds = creds._set_blocking_regional_access_boundary_lookup()
 
             request = mock.Mock()
             headers = {}