Revert "Revert "Revert "Revert "File watcher authorization policy pro…

…vider implementation"" (grpc#27605)" (grpc#27644)" (grpc#27645)

This reverts commit b8e01f7.

BUILD

@@ -3096,6 +3096,7 @@ grpc_cc_library(
     language = "c++",
     deps = [
         "gpr_base",
+        "grpc_base",
         "grpc_matchers",
         "grpc_rbac_engine",
         "grpc_secure",

include/grpc/grpc_security.h

@@ -1109,6 +1109,24 @@ grpc_authorization_policy_provider_static_data_create(
     const char* authz_policy, grpc_status_code* code,
     const char** error_details);
 
+/**
+ * EXPERIMENTAL - Subject to change.
+ * Creates a grpc_authorization_policy_provider by watching for SDK
+ * authorization policy changes in filesystem.
+ * - authz_policy is the file path of SDK authorization policy.
+ * - refresh_interval_sec is the amount of time the internal thread would wait
+ *   before checking for file updates.
+ * - code is the error status code on failure. On success, it equals
+ *   GRPC_STATUS_OK.
+ * - error_details contains details about the error if any. If the
+ *   initialization is successful, it will be null. Caller must use gpr_free to
+ *   destroy this string.
+ */
+GRPCAPI grpc_authorization_policy_provider*
+grpc_authorization_policy_provider_file_watcher_create(
+    const char* authz_policy_path, unsigned int refresh_interval_sec,
+    grpc_status_code* code, const char** error_details);
+
 /**
  * EXPERIMENTAL - Subject to change.
  * Releases grpc_authorization_policy_provider object. The creator of

include/grpcpp/security/authorization_policy_provider.h

@@ -20,7 +20,7 @@
 #include <grpc/status.h>
 #include <grpcpp/impl/codegen/grpc_library.h>
 
-// TODO(yihuazhang): remove the forward declaration here and include
+// TODO(yihuazhang): remove the forward declarations here and include
 // <grpc/grpc_security.h> directly once the insecure builds are cleaned up.
 typedef struct grpc_authorization_policy_provider
     grpc_authorization_policy_provider;
@@ -61,6 +61,31 @@ class StaticDataAuthorizationPolicyProvider
   grpc_authorization_policy_provider* c_provider_ = nullptr;
 };
 
+// Implementation obtains authorization policy by watching for changes in
+// filesystem.
+class FileWatcherAuthorizationPolicyProvider
+    : public AuthorizationPolicyProviderInterface {
+ public:
+  static std::shared_ptr<FileWatcherAuthorizationPolicyProvider> Create(
+      const std::string& authz_policy_path, unsigned int refresh_interval_sec,
+      grpc::Status* status);
+
+  // Use factory method "Create" to create an instance of
+  // FileWatcherAuthorizationPolicyProvider.
+  explicit FileWatcherAuthorizationPolicyProvider(
+      grpc_authorization_policy_provider* provider)
+      : c_provider_(provider) {}
+
+  ~FileWatcherAuthorizationPolicyProvider() override;
+
+  grpc_authorization_policy_provider* c_provider() override {
+    return c_provider_;
+  }
+
+ private:
+  grpc_authorization_policy_provider* c_provider_ = nullptr;
+};
+
 }  // namespace experimental
 }  // namespace grpc
 

src/core/lib/security/authorization/authorization_policy_provider.h

@@ -27,7 +27,7 @@ struct grpc_authorization_policy_provider
     grpc_core::RefCountedPtr<grpc_core::AuthorizationEngine> allow_engine;
     grpc_core::RefCountedPtr<grpc_core::AuthorizationEngine> deny_engine;
   };
-  virtual AuthorizationEngines engines() const = 0;
+  virtual AuthorizationEngines engines() = 0;
 };
 
 #endif  // GRPC_CORE_LIB_SECURITY_AUTHORIZATION_AUTHORIZATION_POLICY_PROVIDER_H

src/core/lib/security/authorization/grpc_authorization_engine.h

@@ -38,6 +38,9 @@ class GrpcAuthorizationEngine : public AuthorizationEngine {
 
   Rbac::Action action() { return action_; }
 
+  // Required only for testing purpose.
+  size_t num_policies() { return policies_.size(); }
+
   // Evaluates incoming request against RBAC policy and makes a decision to
   // whether allow/deny this request.
   Decision Evaluate(const EvaluateArgs& args) const override;

src/core/lib/security/authorization/grpc_authorization_policy_provider.cc

@@ -19,10 +19,14 @@
 #include <grpc/grpc_security.h>
 #include <grpc/support/string_util.h>
 
+#include "src/core/lib/iomgr/load_file.h"
 #include "src/core/lib/security/authorization/grpc_authorization_engine.h"
+#include "src/core/lib/slice/slice_internal.h"
 
 namespace grpc_core {
 
+extern TraceFlag grpc_sdk_authz_trace;
+
 absl::StatusOr<RefCountedPtr<grpc_authorization_policy_provider>>
 StaticDataAuthorizationPolicyProvider::Create(absl::string_view authz_policy) {
   auto policies_or = GenerateRbacPolicies(authz_policy);
@@ -40,6 +44,113 @@ StaticDataAuthorizationPolicyProvider::StaticDataAuthorizationPolicyProvider(
       deny_engine_(MakeRefCounted<GrpcAuthorizationEngine>(
           std::move(policies.deny_policy))) {}
 
+namespace {
+
+absl::StatusOr<std::string> ReadPolicyFromFile(absl::string_view policy_path) {
+  grpc_slice policy_slice = grpc_empty_slice();
+  grpc_error_handle error =
+      grpc_load_file(std::string(policy_path).c_str(), 0, &policy_slice);
+  if (error != GRPC_ERROR_NONE) {
+    absl::Status status =
+        absl::InvalidArgumentError(grpc_error_std_string(error));
+    GRPC_ERROR_UNREF(error);
+    return status;
+  }
+  std::string policy_contents(StringViewFromSlice(policy_slice));
+  grpc_slice_unref_internal(policy_slice);
+  return policy_contents;
+}
+
+gpr_timespec TimeoutSecondsToDeadline(int64_t seconds) {
+  return gpr_time_add(gpr_now(GPR_CLOCK_MONOTONIC),
+                      gpr_time_from_seconds(seconds, GPR_TIMESPAN));
+}
+
+}  // namespace
+
+absl::StatusOr<RefCountedPtr<grpc_authorization_policy_provider>>
+FileWatcherAuthorizationPolicyProvider::Create(
+    absl::string_view authz_policy_path, unsigned int refresh_interval_sec) {
+  GPR_ASSERT(!authz_policy_path.empty());
+  GPR_ASSERT(refresh_interval_sec > 0);
+  absl::Status status;
+  auto provider = MakeRefCounted<FileWatcherAuthorizationPolicyProvider>(
+      authz_policy_path, refresh_interval_sec, &status);
+  if (!status.ok()) return status;
+  return provider;
+}
+
+FileWatcherAuthorizationPolicyProvider::FileWatcherAuthorizationPolicyProvider(
+    absl::string_view authz_policy_path, unsigned int refresh_interval_sec,
+    absl::Status* status)
+    : authz_policy_path_(std::string(authz_policy_path)),
+      refresh_interval_sec_(refresh_interval_sec) {
+  gpr_event_init(&shutdown_event_);
+  // Initial read is done synchronously.
+  *status = ForceUpdate();
+  if (!status->ok()) {
+    return;
+  }
+  auto thread_lambda = [](void* arg) {
+    WeakRefCountedPtr<FileWatcherAuthorizationPolicyProvider> provider(
+        static_cast<FileWatcherAuthorizationPolicyProvider*>(arg));
+    GPR_ASSERT(provider != nullptr);
+    while (true) {
+      void* value = gpr_event_wait(
+          &provider->shutdown_event_,
+          TimeoutSecondsToDeadline(provider->refresh_interval_sec_));
+      if (value != nullptr) {
+        return;
+      }
+      absl::Status status = provider->ForceUpdate();
+      if (GRPC_TRACE_FLAG_ENABLED(grpc_sdk_authz_trace) && !status.ok()) {
+        gpr_log(GPR_ERROR,
+                "authorization policy reload status. code=%d error_details=%s",
+                status.code(), std::string(status.message()).c_str());
+      }
+    }
+  };
+  refresh_thread_ = absl::make_unique<grpc_core::Thread>(
+      "FileWatcherAuthorizationPolicyProvider_refreshing_thread", thread_lambda,
+      WeakRef().release());
+  refresh_thread_->Start();
+}
+
+absl::Status FileWatcherAuthorizationPolicyProvider::ForceUpdate() {
+  absl::StatusOr<std::string> file_contents =
+      ReadPolicyFromFile(authz_policy_path_);
+  if (!file_contents.ok()) {
+    return file_contents.status();
+  }
+  if (file_contents_ == *file_contents) {
+    return absl::OkStatus();
+  }
+  file_contents_ = std::move(*file_contents);
+  auto rbac_policies_or = GenerateRbacPolicies(file_contents_);
+  if (!rbac_policies_or.ok()) {
+    return rbac_policies_or.status();
+  }
+  grpc_core::MutexLock lock(&mu_);
+  allow_engine_ = MakeRefCounted<GrpcAuthorizationEngine>(
+      std::move(rbac_policies_or->allow_policy));
+  deny_engine_ = MakeRefCounted<GrpcAuthorizationEngine>(
+      std::move(rbac_policies_or->deny_policy));
+  if (GRPC_TRACE_FLAG_ENABLED(grpc_sdk_authz_trace)) {
+    gpr_log(GPR_INFO,
+            "authorization policy reload status: successfully loaded new "
+            "policy\n%s",
+            file_contents_.c_str());
+  }
+  return absl::OkStatus();
+}
+
+void FileWatcherAuthorizationPolicyProvider::Orphan() {
+  gpr_event_set(&shutdown_event_, reinterpret_cast<void*>(1));
+  if (refresh_thread_ != nullptr) {
+    refresh_thread_->Join();
+  }
+}
+
 }  // namespace grpc_core
 
 // Wrapper APIs declared in grpc_security.h
@@ -57,8 +168,22 @@ grpc_authorization_policy_provider_static_data_create(
         gpr_strdup(std::string(provider_or.status().message()).c_str());
     return nullptr;
   }
-  *code = GRPC_STATUS_OK;
-  *error_details = nullptr;
+  return provider_or->release();
+}
+
+grpc_authorization_policy_provider*
+grpc_authorization_policy_provider_file_watcher_create(
+    const char* authz_policy_path, unsigned int refresh_interval_sec,
+    grpc_status_code* code, const char** error_details) {
+  GPR_ASSERT(authz_policy_path != nullptr);
+  auto provider_or = grpc_core::FileWatcherAuthorizationPolicyProvider::Create(
+      authz_policy_path, refresh_interval_sec);
+  if (!provider_or.ok()) {
+    *code = static_cast<grpc_status_code>(provider_or.status().code());
+    *error_details =
+        gpr_strdup(std::string(provider_or.status().message()).c_str());
+    return nullptr;
+  }
   return provider_or->release();
 }
 

src/core/lib/security/authorization/grpc_authorization_policy_provider.h

@@ -21,6 +21,8 @@
 
 #include "absl/status/statusor.h"
 
+#include "src/core/lib/gprpp/sync.h"
+#include "src/core/lib/gprpp/thd.h"
 #include "src/core/lib/security/authorization/authorization_policy_provider.h"
 #include "src/core/lib/security/authorization/rbac_translator.h"
 
@@ -36,9 +38,11 @@ class StaticDataAuthorizationPolicyProvider
   static absl::StatusOr<RefCountedPtr<grpc_authorization_policy_provider>>
   Create(absl::string_view authz_policy);
 
+  // Use factory method "Create" to create an instance of
+  // StaticDataAuthorizationPolicyProvider.
   explicit StaticDataAuthorizationPolicyProvider(RbacPolicies policies);
 
-  AuthorizationEngines engines() const override {
+  AuthorizationEngines engines() override {
     return {allow_engine_, deny_engine_};
   }
 
@@ -49,8 +53,50 @@ class StaticDataAuthorizationPolicyProvider
   RefCountedPtr<AuthorizationEngine> deny_engine_;
 };
 
-// TODO(ashithasantosh): Add implementation for file watcher authorization
-// policy provider.
+// Provider class will get SDK Authorization policy from provided file path.
+// This policy will be translated to Envoy RBAC policies and used to initialize
+// allow and deny AuthorizationEngine objects. This provider will periodically
+// load file contents in specified path, and upon modification update the engine
+// instances with new policy configuration. During reload if the file contents
+// are invalid or there are I/O errors, we will skip that particular update and
+// log error status. The authorization decisions will be made using the latest
+// valid policy.
+class FileWatcherAuthorizationPolicyProvider
+    : public grpc_authorization_policy_provider {
+ public:
+  static absl::StatusOr<RefCountedPtr<grpc_authorization_policy_provider>>
+  Create(absl::string_view authz_policy_path,
+         unsigned int refresh_interval_sec);
+
+  // Use factory method "Create" to create an instance of
+  // FileWatcherAuthorizationPolicyProvider.
+  FileWatcherAuthorizationPolicyProvider(absl::string_view authz_policy_path,
+                                         unsigned int refresh_interval_sec,
+                                         absl::Status* status);
+
+  void Orphan() override;
+
+  AuthorizationEngines engines() override {
+    grpc_core::MutexLock lock(&mu_);
+    return {allow_engine_, deny_engine_};
+  }
+
+ private:
+  // Force an update from the file system regardless of the interval.
+  absl::Status ForceUpdate();
+
+  std::string authz_policy_path_;
+  std::string file_contents_;
+  unsigned int refresh_interval_sec_;
+
+  std::unique_ptr<Thread> refresh_thread_;
+  gpr_event shutdown_event_;
+
+  grpc_core::Mutex mu_;
+  // Engines created using authz_policy_.
+  RefCountedPtr<AuthorizationEngine> allow_engine_ ABSL_GUARDED_BY(mu_);
+  RefCountedPtr<AuthorizationEngine> deny_engine_ ABSL_GUARDED_BY(mu_);
+};
 
 }  // namespace grpc_core
 

src/cpp/server/authorization_policy_provider.cc

@@ -14,6 +14,7 @@
 
 #include <grpc/grpc_security.h>
 #include <grpc/support/alloc.h>
+#include <grpc/support/log.h>
 #include <grpcpp/security/authorization_policy_provider.h>
 
 namespace grpc {
@@ -22,7 +23,7 @@ namespace experimental {
 std::shared_ptr<StaticDataAuthorizationPolicyProvider>
 StaticDataAuthorizationPolicyProvider::Create(const std::string& authz_policy,
                                               grpc::Status* status) {
-  grpc_status_code code;
+  grpc_status_code code = GRPC_STATUS_OK;
   const char* error_details;
   grpc_authorization_policy_provider* provider =
       grpc_authorization_policy_provider_static_data_create(
@@ -41,5 +42,28 @@ StaticDataAuthorizationPolicyProvider::
   grpc_authorization_policy_provider_release(c_provider_);
 }
 
+std::shared_ptr<FileWatcherAuthorizationPolicyProvider>
+FileWatcherAuthorizationPolicyProvider::Create(
+    const std::string& authz_policy_path, unsigned int refresh_interval_sec,
+    grpc::Status* status) {
+  grpc_status_code code = GRPC_STATUS_OK;
+  const char* error_details;
+  grpc_authorization_policy_provider* provider =
+      grpc_authorization_policy_provider_file_watcher_create(
+          authz_policy_path.c_str(), refresh_interval_sec, &code,
+          &error_details);
+  if (code != GRPC_STATUS_OK) {
+    *status = grpc::Status(static_cast<grpc::StatusCode>(code), error_details);
+    gpr_free(const_cast<char*>(error_details));
+    return nullptr;
+  }
+  return std::make_shared<FileWatcherAuthorizationPolicyProvider>(provider);
+}
+
+FileWatcherAuthorizationPolicyProvider::
+    ~FileWatcherAuthorizationPolicyProvider() {
+  grpc_authorization_policy_provider_release(c_provider_);
+}
+
 }  // namespace experimental
 }  // namespace grpc