fix(storage): scopes should disable self-signed JWTs

The storage service does not support self-signed JWTs with scopes. With
this change, self-signed JWTs will be automatically disabled when using
the legacy `storage::oauth2::Credentials`.

google/cloud/internal/oauth2_service_account_credentials.cc

@@ -80,8 +80,9 @@ StatusOr<ServiceAccountCredentialsInfo> ParseServiceAccountCredentials(
       // "token_uri" attribute in the JSON object.  In this case, we try using
       // the default value.
       credentials.value(token_uri_key, default_token_uri),
-      /*scopes*/ {},
-      /*subject*/ {}};
+      /*scopes=*/absl::nullopt,
+      /*subject=*/absl::nullopt,
+      /*enable_self_signed_jwt=*/true};
 }
 
 std::pair<std::string, std::string> AssertionComponentsFromInfo(
@@ -332,13 +333,17 @@ StatusOr<ServiceAccountCredentialsInfo> ParseServiceAccountP12File(
                                        kP12PrivateKeyIdMarker,
                                        std::move(private_key),
                                        GoogleOAuthRefreshEndpoint(),
-                                       /*scopes*/ {},
-                                       /*subject*/ {}};
+                                       /*scopes=*/{},
+                                       /*subject=*/{},
+                                       /*enable_self_signed_jwt=*/false};
 }
 #include "google/cloud/internal/diagnostics_pop.inc"
 
 bool ServiceAccountUseOAuth(ServiceAccountCredentialsInfo const& info) {
-  if (info.private_key_id == kP12PrivateKeyIdMarker) return true;
+  if (info.private_key_id == kP12PrivateKeyIdMarker ||
+      !info.enable_self_signed_jwt) {
+    return true;
+  }
   auto disable_jwt = google::cloud::internal::GetEnv(
       "GOOGLE_CLOUD_CPP_EXPERIMENTAL_DISABLE_SELF_SIGNED_JWT");
   return disable_jwt.has_value();

google/cloud/internal/oauth2_service_account_credentials.h

@@ -49,6 +49,7 @@ struct ServiceAccountCredentialsInfo {
   absl::optional<std::set<std::string>> scopes;
   // See https://developers.google.com/identity/protocols/OAuth2ServiceAccount.
   absl::optional<std::string> subject;
+  bool enable_self_signed_jwt;
 };
 
 /// Indicates whether or not to use a self-signed JWT or issue a request to

google/cloud/storage/internal/unified_rest_credentials.cc

@@ -61,18 +61,6 @@ class WrapRestCredentials : public oauth2::Credentials {
   std::shared_ptr<oauth2_internal::Credentials> impl_;
 };
 
-oauth2_internal::ServiceAccountCredentialsInfo MapInfo(
-    oauth2::ServiceAccountCredentialsInfo info) {
-  oauth2_internal::ServiceAccountCredentialsInfo result;
-  result.client_email = std::move(info.client_email);
-  result.private_key_id = std::move(info.private_key_id);
-  result.private_key = std::move(info.private_key);
-  result.token_uri = std::move(info.token_uri);
-  result.scopes = std::move(info.scopes);
-  result.subject = std::move(info.subject);
-  return result;
-}
-
 struct RestVisitor : public CredentialsVisitor {
   std::shared_ptr<oauth2::Credentials> result;
 
@@ -102,7 +90,7 @@ struct RestVisitor : public CredentialsVisitor {
     }
     result = std::make_shared<WrapRestCredentials>(
         std::make_shared<oauth2_internal::ServiceAccountCredentials>(
-            MapInfo(*std::move(info))));
+            internal::MapServiceAccountCredentialsInfo(*std::move(info))));
   }
 };
 

google/cloud/storage/oauth2/service_account_credentials.cc

@@ -56,10 +56,7 @@ std::pair<std::string, std::string> AssertionComponentsFromInfo(
     ServiceAccountCredentialsInfo const& info,
     std::chrono::system_clock::time_point now) {
   return oauth2_internal::AssertionComponentsFromInfo(
-      oauth2_internal::ServiceAccountCredentialsInfo{
-          info.client_email, info.private_key_id, info.private_key,
-          info.token_uri, info.scopes, info.subject},
-      now);
+      internal::MapServiceAccountCredentialsInfo(info), now);
 }
 
 std::string MakeJWTAssertion(std::string const& header,
@@ -73,10 +70,7 @@ std::string CreateServiceAccountRefreshPayload(
     ServiceAccountCredentialsInfo const& info, std::string const&,
     std::chrono::system_clock::time_point now) {
   auto params = oauth2_internal::CreateServiceAccountRefreshPayload(
-      oauth2_internal::ServiceAccountCredentialsInfo{
-          info.client_email, info.private_key_id, info.private_key,
-          info.token_uri, info.scopes, info.subject},
-      now);
+      internal::MapServiceAccountCredentialsInfo(info), now);
   return absl::StrJoin(params, "&", absl::PairFormatter("="));
 }
 
@@ -111,26 +105,44 @@ ParseServiceAccountRefreshResponse(
 StatusOr<std::string> MakeSelfSignedJWT(
     ServiceAccountCredentialsInfo const& info,
     std::chrono::system_clock::time_point tp) {
-  // This only runs about once an hour, the copies are ugly, but should be
-  // harmless.
-  oauth2_internal::ServiceAccountCredentialsInfo mapped;
-  mapped.client_email = info.client_email;
-  mapped.private_key_id = info.private_key_id;
-  mapped.private_key = info.private_key;
-  mapped.token_uri = info.token_uri;
-  mapped.scopes = info.scopes;
-  mapped.subject = info.subject;
+  auto mapped = internal::MapServiceAccountCredentialsInfo(info);
   return ::google::cloud::oauth2_internal::MakeSelfSignedJWT(mapped, tp);
 }
 
 bool ServiceAccountUseOAuth(ServiceAccountCredentialsInfo const& info) {
   if (info.private_key_id == kP12PrivateKeyIdMarker) return true;
+  // Self-signed JWTs do not work in GCS if they have scopes.
+  if (info.scopes.has_value()) return true;
   auto disable_jwt = google::cloud::internal::GetEnv(
       "GOOGLE_CLOUD_CPP_EXPERIMENTAL_DISABLE_SELF_SIGNED_JWT");
   return disable_jwt.has_value();
 }
 
+ServiceAccountCredentials<storage::internal::CurlRequestBuilder,
+                          std::chrono::system_clock>::
+    ServiceAccountCredentials(ServiceAccountCredentialsInfo info,
+                              ChannelOptions const& options)
+    : impl_(absl::make_unique<oauth2_internal::ServiceAccountCredentials>(
+          internal::MapServiceAccountCredentialsInfo(std::move(info)),
+          Options{}.set<CARootsFilePathOption>(options.ssl_root_path()))) {}
+
 }  // namespace oauth2
+
+namespace internal {
+
+oauth2_internal::ServiceAccountCredentialsInfo MapServiceAccountCredentialsInfo(
+    oauth2::ServiceAccountCredentialsInfo info) {
+  // Storage has more stringent requirements w.r.t. self-signed JWTs
+  // than most services (which the base class
+  // Disable them in the implementation class
+  auto enable_self_signed_jwt = !ServiceAccountUseOAuth(info);
+  return {std::move(info.client_email), std::move(info.private_key_id),
+          std::move(info.private_key),  std::move(info.token_uri),
+          std::move(info.scopes),       std::move(info.subject),
+          enable_self_signed_jwt};
+}
+
+}  // namespace internal
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
 }  // namespace storage
 }  // namespace cloud

google/cloud/storage/oauth2/service_account_credentials.h

@@ -208,15 +208,10 @@ class ServiceAccountCredentials<storage::internal::CurlRequestBuilder,
                                 std::chrono::system_clock>
     : public Credentials {
  public:
-  explicit ServiceAccountCredentials(ServiceAccountCredentialsInfo const& info)
+  explicit ServiceAccountCredentials(ServiceAccountCredentialsInfo info)
       : ServiceAccountCredentials(std::move(info), {}) {}
-  ServiceAccountCredentials(ServiceAccountCredentialsInfo const& info,
-                            ChannelOptions const& options)
-      : impl_(absl::make_unique<oauth2_internal::ServiceAccountCredentials>(
-            oauth2_internal::ServiceAccountCredentialsInfo{
-                info.client_email, info.private_key_id, info.private_key,
-                info.token_uri, info.scopes, info.subject},
-            Options{}.set<CARootsFilePathOption>(options.ssl_root_path()))) {}
+  ServiceAccountCredentials(ServiceAccountCredentialsInfo info,
+                            ChannelOptions const& options);
 
   StatusOr<std::string> AuthorizationHeader() override {
     return oauth2_internal::AuthorizationHeaderJoined(*impl_);
@@ -344,6 +339,13 @@ class ServiceAccountCredentials : public Credentials {
 };
 
 }  // namespace oauth2
+
+namespace internal {
+
+oauth2_internal::ServiceAccountCredentialsInfo MapServiceAccountCredentialsInfo(
+    oauth2::ServiceAccountCredentialsInfo info);
+
+}  // namespace internal
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
 }  // namespace storage
 }  // namespace cloud

google/cloud/storage/oauth2/service_account_credentials_test.cc

@@ -782,24 +782,36 @@ TEST_F(ServiceAccountCredentialsTest, UseOauth) {
 
   auto p12_info = ParseServiceAccountP12File(filename);
   EXPECT_EQ(0, std::remove(filename.c_str()));
-  if (!p12_info) GTEST_SKIP();  // Some environments do not support PKCS#12
-  ASSERT_STATUS_OK(p12_info);
 
-  auto json_info = ParseServiceAccountCredentials(MakeTestContents(), "test");
-  ASSERT_STATUS_OK(json_info);
+  auto json_info_without_scopes =
+      ParseServiceAccountCredentials(MakeTestContents(), "test");
+  ASSERT_STATUS_OK(json_info_without_scopes);
+  ASSERT_FALSE(json_info_without_scopes->scopes.has_value());
+  auto json_info_with_scopes = *json_info_without_scopes;
+  json_info_with_scopes.scopes =
+      std::set<std::string>{{GoogleOAuthScopeCloudPlatform()}};
 
   struct TestCase {
     std::string name;
     ServiceAccountCredentialsInfo info;
     absl::optional<std::string> environment;
     bool expected;
-  } cases[] = {
-      {"JSON/no-env", *json_info, absl::nullopt, false},
-      {"JSON/env", *json_info, "1", true},
-      {"P12/no-env", *p12_info, absl::nullopt, true},
-      {"P12/env", *p12_info, "1", true},
   };
 
+  std::vector<TestCase> cases = {
+      {"JSON/with-scopes/no-env", json_info_with_scopes, absl::nullopt, true},
+      {"JSON/with-scopes/env", json_info_with_scopes, "1", true},
+      {"JSON/no-scopes/no-env", *json_info_without_scopes, absl::nullopt,
+       false},
+      {"JSON/no-scopes/env", *json_info_without_scopes, "1", true},
+  };
+  if (p12_info) {
+    // Some environments do not support PKCS$12, we need to test the other
+    // cases.
+    cases.push_back({"P12/no-env", *p12_info, absl::nullopt, true});
+    cases.push_back({"P12/env", *p12_info, "1", true});
+  }
+
   for (auto const& test : cases) {
     SCOPED_TRACE("Testing for " + test.name);
     ScopedEnvironment env(