jwt_authn: support fetching jwks in the background

api/envoy/extensions/filters/http/jwt_authn/v3/config.proto

@@ -232,6 +232,35 @@ message RemoteJwks {
   // Duration after which the cached JWKS should be expired. If not specified, default cache
   // duration is 5 minutes.
   google.protobuf.Duration cache_duration = 2;
+
+  // Fetch Jwks asynchronously in the main thread before the listener is activated.
+  // Fetched Jwks can be used by all worker threads.
+  //
+  // If this feature is not enabled:
+  //
+  // * The Jwks is fetched on-demand when the requests come. During the fetching, first
+  //   few requests are paused until the Jwks is fetched.
+  // * Each worker thread fetches its own Jwks since Jwks cache is per worker thread.
+  //
+  // If this feature is enabled:
+  //
+  // * Fetched Jwks is done in the main thread before the listener is activated. Its fetched
+  //   Jwks can be used by all worker threads. Each worker thread doesn't need to fetch its own.
+  // * Jwks is ready when the requests come, not need to wait for the Jwks fetching.
+  //
+  JwksAsyncFetch async_fetch = 3;
+}
+
+// Fetch Jwks asynchronously in the main thread when the filter config is parsed.
+// The listener is activated only after the Jwks is fetched.
+// When the Jwks is expired in the cache, it is fetched again in the main thread.
+// The fetched Jwks from the main thread can be used by all worker threads.
+message JwksAsyncFetch {
+  // If false, the listener is activated after the initial fetch is completed.
+  // The initial fetch result can be either successful or failed.
+  // If true, it is activated without waiting for the initial fetch to complete.
+  // Default is false.
+  bool fast_listener = 1;
 }
 
 // This message specifies a header location to extract JWT token.

docs/root/version_history/current.rst

@@ -76,6 +76,7 @@ New Features
   :ref:`xff <envoy_v3_api_msg_extensions.http.original_ip_detection.xff.v3.XffConfig>` extension.
 * http: added the ability to :ref:`unescape slash sequences<envoy_v3_api_field_extensions.filters.network.http_connection_manager.v3.HttpConnectionManager.path_with_escaped_slashes_action>` in the path. Requests with unescaped slashes can be proxied, rejected or redirected to the new unescaped path. By default this feature is disabled. The default behavior can be overridden through :ref:`http_connection_manager.path_with_escaped_slashes_action<config_http_conn_man_runtime_path_with_escaped_slashes_action>` runtime variable. This action can be selectively enabled for a portion of requests by setting the :ref:`http_connection_manager.path_with_escaped_slashes_action_sampling<config_http_conn_man_runtime_path_with_escaped_slashes_action_enabled>` runtime variable.
 * http: added upstream and downstream alpha HTTP/3 support! See :ref:`quic_options <envoy_v3_api_field_config.listener.v3.UdpListenerConfig.quic_options>` for downstream and the new http3_protocol_options in :ref:`http_protocol_options <envoy_v3_api_msg_extensions.upstreams.http.v3.HttpProtocolOptions>` for upstream HTTP/3.
+* jwt_authn: added support to fetch remote jwks asynchronously specified by :ref:`async_fetch <envoy_v3_api_field_extensions.filters.http.jwt_authn.v3.RemoteJwks.async_fetch>`.
 * listener: added ability to change an existing listener's address.
 * local_rate_limit_filter: added suppoort for locally rate limiting http requests on a per connection basis. This can be enabled by setting the :ref:`local_rate_limit_per_downstream_connection <envoy_v3_api_field_extensions.filters.http.local_ratelimit.v3.LocalRateLimit.local_rate_limit_per_downstream_connection>` field to true.
 * metric service: added support for sending metric tags as labels. This can be enabled by setting the :ref:`emit_tags_as_labels <envoy_v3_api_field_config.metrics.v3.MetricsServiceConfig.emit_tags_as_labels>` field to true.

source/extensions/filters/http/jwt_authn/BUILD

@@ -20,6 +20,33 @@ envoy_cc_library(
     ],
 )
 
+envoy_cc_library(
+    name = "stats_lib",
+    hdrs = ["stats.h"],
+    deps = [
+        "//include/envoy/stats:stats_macros",
+    ],
+)
+
+envoy_cc_library(
+    name = "jwks_async_fetcher_lib",
+    srcs = ["jwks_async_fetcher.cc"],
+    hdrs = ["jwks_async_fetcher.h"],
+    external_deps = [
+        "jwt_verify_lib",
+    ],
+    deps = [
+        ":stats_lib",
+        "//include/envoy/server:factory_context_interface",
+        "//source/common/common:minimal_logger_lib",
+        "//source/common/init:target_lib",
+        "//source/common/protobuf:utility_lib",
+        "//source/common/tracing:http_tracer_lib",
+        "//source/extensions/filters/http/common:jwks_fetcher_lib",
+        "@envoy_api//envoy/extensions/filters/http/jwt_authn/v3:pkg_cc_proto",
+    ],
+)
+
 envoy_cc_library(
     name = "jwks_cache_lib",
     srcs = ["jwks_cache.cc"],
@@ -28,9 +55,8 @@ envoy_cc_library(
         "jwt_verify_lib",
     ],
     deps = [
-        "//source/common/common:minimal_logger_lib",
+        "jwks_async_fetcher_lib",
         "//source/common/config:datasource_lib",
-        "//source/common/protobuf:utility_lib",
         "@envoy_api//envoy/extensions/filters/http/jwt_authn/v3:pkg_cc_proto",
     ],
 )
@@ -58,7 +84,7 @@ envoy_cc_library(
         "jwt_verify_lib",
     ],
     deps = [
-        ":filter_config_interface",
+        ":filter_config_lib",
         ":matchers_lib",
         "//include/envoy/http:filter_interface",
         "//source/common/http:headers_lib",
@@ -109,7 +135,7 @@ envoy_cc_library(
 )
 
 envoy_cc_library(
-    name = "filter_config_interface",
+    name = "filter_config_lib",
     srcs = ["filter_config.cc"],
     hdrs = ["filter_config.h"],
     deps = [

source/extensions/filters/http/jwt_authn/authenticator.cc

@@ -221,6 +221,7 @@ void AuthenticatorImpl::startVerify() {
 }
 
 void AuthenticatorImpl::onJwksSuccess(google::jwt_verify::JwksPtr&& jwks) {
+  jwks_cache_.stats().jwks_fetch_success_.inc();
   const Status status = jwks_data_->setRemoteJwks(std::move(jwks))->getStatus();
   if (status != Status::Ok) {
     doneWithStatus(status);
@@ -229,7 +230,10 @@ void AuthenticatorImpl::onJwksSuccess(google::jwt_verify::JwksPtr&& jwks) {
   }
 }
 
-void AuthenticatorImpl::onJwksError(Failure) { doneWithStatus(Status::JwksFetchFail); }
+void AuthenticatorImpl::onJwksError(Failure) {
+  jwks_cache_.stats().jwks_fetch_failed_.inc();
+  doneWithStatus(Status::JwksFetchFail);
+}
 
 void AuthenticatorImpl::onDestroy() {
   if (fetcher_) {

source/extensions/filters/http/jwt_authn/authenticator.h

@@ -2,7 +2,6 @@
 
 #include "envoy/server/filter_config.h"
 
-#include "extensions/filters/http/common/jwks_fetcher.h"
 #include "extensions/filters/http/jwt_authn/extractor.h"
 #include "extensions/filters/http/jwt_authn/jwks_cache.h"
 
@@ -21,11 +20,6 @@ using AuthenticatorCallback = std::function<void(const ::google::jwt_verify::Sta
 
 using SetPayloadCallback = std::function<void(const std::string&, const ProtobufWkt::Struct&)>;
 
-/**
- *  CreateJwksFetcherCb is a callback interface for creating a JwksFetcher instance.
- */
-using CreateJwksFetcherCb = std::function<Common::JwksFetcherPtr(Upstream::ClusterManager&)>;
-
 /**
  *  Authenticator object to handle all JWT authentication flow.
  */

source/extensions/filters/http/jwt_authn/filter_config.cc

@@ -19,8 +19,7 @@ FilterConfigImpl::FilterConfigImpl(
 
   ENVOY_LOG(debug, "Loaded JwtAuthConfig: {}", proto_config_.DebugString());
 
-  jwks_cache_ =
-      JwksCache::create(proto_config_, time_source_, context.api(), context.threadLocal());
+  jwks_cache_ = JwksCache::create(proto_config_, context, Common::JwksFetcher::create, stats_);
 
   std::vector<std::string> names;
   for (const auto& it : proto_config_.requirement_map()) {

source/extensions/filters/http/jwt_authn/filter_config.h

@@ -9,6 +9,7 @@
 #include "envoy/thread_local/thread_local.h"
 
 #include "extensions/filters/http/jwt_authn/matcher.h"
+#include "extensions/filters/http/jwt_authn/stats.h"
 #include "extensions/filters/http/jwt_authn/verifier.h"
 
 #include "absl/container/flat_hash_map.h"
@@ -18,21 +19,6 @@ namespace Extensions {
 namespace HttpFilters {
 namespace JwtAuthn {
 
-/**
- * All stats for the Jwt Authn filter. @see stats_macros.h
- */
-#define ALL_JWT_AUTHN_FILTER_STATS(COUNTER)                                                        \
-  COUNTER(allowed)                                                                                 \
-  COUNTER(cors_preflight_bypassed)                                                                 \
-  COUNTER(denied)
-
-/**
- * Wrapper struct for jwt_authn filter stats. @see stats_macros.h
- */
-struct JwtAuthnFilterStats {
-  ALL_JWT_AUTHN_FILTER_STATS(GENERATE_COUNTER_STRUCT)
-};
-
 /**
  * The per-route filter config
  */

source/extensions/filters/http/jwt_authn/jwks_async_fetcher.cc

@@ -0,0 +1,98 @@
+#include "extensions/filters/http/jwt_authn/jwks_async_fetcher.h"
+
+#include "common/protobuf/utility.h"
+#include "common/tracing/http_tracer_impl.h"
+
+using envoy::extensions::filters::http::jwt_authn::v3::RemoteJwks;
+
+namespace Envoy {
+namespace Extensions {
+namespace HttpFilters {
+namespace JwtAuthn {
+namespace {
+
+// Default cache expiration time in 5 minutes.
+constexpr int PubkeyCacheExpirationSec = 600;
+
+} // namespace
+
+JwksAsyncFetcher::JwksAsyncFetcher(const RemoteJwks& remote_jwks,
+                                   Server::Configuration::FactoryContext& context,
+                                   CreateJwksFetcherCb create_fetcher_fn,
+                                   JwtAuthnFilterStats& stats, JwksDoneFetched done_fn)
+    : remote_jwks_(remote_jwks), context_(context), create_fetcher_fn_(create_fetcher_fn),
+      stats_(stats), done_fn_(done_fn), cache_duration_(getCacheDuration(remote_jwks)),
+      debug_name_(absl::StrCat("Jwks async fetching url=", remote_jwks_.http_uri().uri())) {
+  // if async_fetch is not enabled, do nothing.
+  if (!remote_jwks_.has_async_fetch()) {
+    return;
+  }
+
+  cache_duration_timer_ = context_.dispatcher().createTimer([this]() -> void { fetch(); });
+
+  // For fast_listener, just trigger a fetch, not register with init_manager.
+  if (remote_jwks_.async_fetch().fast_listener()) {
+    fetch();
+    return;
+  }
+
+  // Register to init_manager, force the listener to wait for the fetching.
+  init_target_ = std::make_unique<Init::TargetImpl>(debug_name_, [this]() -> void { fetch(); });
+  context_.initManager().add(*init_target_);
+}
+
+std::chrono::seconds JwksAsyncFetcher::getCacheDuration(const RemoteJwks& remote_jwks) {
+  if (remote_jwks.has_cache_duration()) {
+    return std::chrono::seconds(DurationUtil::durationToSeconds(remote_jwks.cache_duration()));
+  }
+  return std::chrono::seconds(PubkeyCacheExpirationSec);
+}
+
+void JwksAsyncFetcher::fetch() {
+  if (fetcher_) {
+    fetcher_->cancel();
+  }
+
+  ENVOY_LOG(debug, "{}: started", debug_name_);
+  fetcher_ = create_fetcher_fn_(context_.clusterManager());
+  fetcher_->fetch(remote_jwks_.http_uri(), Tracing::NullSpan::instance(), *this);
+}
+
+void JwksAsyncFetcher::handleFetchDone() {
+  if (init_target_) {
+    init_target_->ready();
+    init_target_.reset();
+  }
+
+  cache_duration_timer_->enableTimer(cache_duration_);
+}
+
+void JwksAsyncFetcher::onJwksSuccess(google::jwt_verify::JwksPtr&& jwks) {
+  stats_.jwks_fetch_success_.inc();
+
+  done_fn_(std::move(jwks));
+  handleFetchDone();
+
+  // Note: not to free fetcher_ within onJwksSuccess or onJwksError function.
+  // They are passed to fetcher_->fetch() and are called by fetcher_ after fetch is done.
+  // After calling these callback functions, fetch_ calls its reset() function.
+  // If fetcher_ is freed by the callback,  calling reset() will crash.
+
+  // Not need to free fetcher_. At the next fetch(), it will be freed with a cancel() call.
+  // The cancel() is needed to cancel the old call before the new one is created.
+  // But it is a no-op if the call is completed.
+}
+
+void JwksAsyncFetcher::onJwksError(Failure) {
+  stats_.jwks_fetch_failed_.inc();
+
+  ENVOY_LOG(warn, "{}: failed", debug_name_);
+  handleFetchDone();
+
+  // Note: not to free fetcher_ in this function. Please see comment at onJwksSuccess.
+}
+
+} // namespace JwtAuthn
+} // namespace HttpFilters
+} // namespace Extensions
+} // namespace Envoy