feat(notifications): add authorization before sending notification

front/lib/front/models/notification.ex

@@ -58,7 +58,7 @@ defmodule Front.Models.Notification do
 
   def create(user_id, org_id, notification_data, _metadata \\ nil) do
     Watchman.benchmark("notifications.create_notification_request.duration", fn ->
-      notification = construct_notification(notification_data)
+      notification = construct_notification(notification_data, user_id)
 
       {:ok, channel} = GRPC.Stub.connect(api_endpoint())
 
@@ -90,7 +90,7 @@ defmodule Front.Models.Notification do
 
   def update(user_id, org_id, notification_data) do
     Watchman.benchmark("notifications.update_notification_request.duration", fn ->
-      notification = construct_notification(notification_data)
+      notification = construct_notification(notification_data, user_id)
 
       {:ok, channel} = GRPC.Stub.connect(api_endpoint())
 
@@ -110,13 +110,13 @@ defmodule Front.Models.Notification do
     end)
   end
 
-  def construct_notification(data) do
+  def construct_notification(data, creator_id) do
     rules =
       data.rules
       |> Enum.map(&construct_rule(&1))
 
     PublicApi.Notification.new(
-      metadata: PublicApi.Notification.Metadata.new(name: data.name),
+      metadata: PublicApi.Notification.Metadata.new(name: data.name, creator_id: creator_id),
       spec: PublicApi.Notification.Spec.new(rules: rules)
     )
   end

front/lib/public_api/semaphore/notifications.v1alpha.pb.ex

@@ -22,14 +22,16 @@ defmodule Semaphore.Notifications.V1alpha.Notification.Metadata do
           name: String.t(),
           id: String.t(),
           create_time: integer,
-          update_time: integer
+          update_time: integer,
+          creator_id: String.t()
         }
-  defstruct [:name, :id, :create_time, :update_time]
+  defstruct [:name, :id, :create_time, :update_time, :creator_id]
 
   field(:name, 1, type: :string)
   field(:id, 2, type: :string)
   field(:create_time, 3, type: :int64)
   field(:update_time, 4, type: :int64)
+  field(:creator_id, 5, type: :string)
 end
 
 defmodule Semaphore.Notifications.V1alpha.Notification.Spec do

notifications/config/config.exs

@@ -13,9 +13,6 @@ config :logger, :console,
 
 import_config "_silent_lager.exs"
 
-config :notifications, rbac_endpoint: "localhost:50052"
-config :notifications, secrethub_endpoint: "localhost:50052" # sobelow_skip ["Config.Secrets"]
-
 config :notifications, ecto_repos: [Notifications.Repo]
 
 import_config "#{config_env()}.exs"

notifications/config/test.exs

@@ -8,4 +8,12 @@ config :junit_formatter,
   include_filename?: true,
   include_file_line?: true
 
+config :notifications, rbac_endpoint: "localhost:50052"
+config :notifications, secrethub_endpoint: "localhost:50052" # sobelow_skip ["Config.Secrets"]
+config :notifications, pipeline_endpoint: "localhost:50052" # sobelow_skip ["Config.Secrets"]
+config :notifications, projecthub_endpoint: "localhost:50052" # sobelow_skip ["Config.Secrets"]
+config :notifications, repo_proxy_endpoint: "localhost:50052" # sobelow_skip ["Config.Secrets"]
+config :notifications, workflow_endpoint: "localhost:50052" # sobelow_skip ["Config.Secrets"]
+config :notifications, organization_endpoint: "localhost:50052" # sobelow_skip ["Config.Secrets"]
+
 config :notifications, Notifications.Repo, pool: Ecto.Adapters.SQL.Sandbox

notifications/lib/notifications/api/internal_api/create.ex

@@ -9,9 +9,10 @@ defmodule Notifications.Api.InternalApi.Create do
 
   def run(req) do
     org_id = req.metadata.org_id
+    creator_id = req.metadata.user_id
 
-    with {:ok, :valid} <- Validator.validate(req.notification),
-         {:ok, n} <- create_notification(org_id, req.notification) do
+    with {:ok, :valid} <- Validator.validate(req.notification, creator_id),
+         {:ok, n} <- create_notification(org_id, creator_id, req.notification) do
       %CreateResponse{notification: Serialization.serialize(n)}
     else
       {:error, :invalid_argument, message} ->
@@ -27,12 +28,13 @@ defmodule Notifications.Api.InternalApi.Create do
     end
   end
 
-  defp create_notification(org_id, notification) do
+  defp create_notification(org_id, creator_id, notification) do
     Repo.transaction(fn ->
       n =
         Models.Notification.new(
           org_id,
           notification.name,
+          creator_id,
           Notifications.Util.Transforms.encode_spec(%{rules: notification.rules})
         )
 

notifications/lib/notifications/api/internal_api/update.ex

@@ -9,10 +9,11 @@ defmodule Notifications.Api.InternalApi.Update do
 
   def run(req) do
     org_id = req.metadata.org_id
+    updater_id = req.metadata.user_id
 
-    with {:ok, :valid} <- Validator.validate(req.notification),
+    with {:ok, :valid} <- Validator.validate(req.notification, updater_id),
          {:ok, n} <- find_by_id_or_name(org_id, req.id, req.name),
-         {:ok, n} <- update_notification(n, req.notification) do
+         {:ok, n} <- update_notification(n, updater_id, req.notification) do
       %UpdateResponse{notification: Serialization.serialize(n)}
     else
       {:error, :invalid_argument, message} ->
@@ -37,11 +38,12 @@ defmodule Notifications.Api.InternalApi.Update do
   # notification: existing row from database
   # apiresource: new notification from the API call that needs to be applied
   #
-  defp update_notification(notification, apiresource) do
+  defp update_notification(notification, updater_id, apiresource) do
     Repo.transaction(fn ->
       changes =
         Model.changeset(notification, %{
           name: apiresource.name,
+          creator_id: updater_id,
           spec: Notifications.Util.Transforms.encode_spec(%{rules: apiresource.rules})
         })
 

notifications/lib/notifications/api/public_api/create.ex

@@ -12,8 +12,8 @@ defmodule Notifications.Api.PublicApi.Create do
     Logger.info("#{inspect(org_id)} #{inspect(user_id)} #{name}")
 
     with {:ok, :authorized} <- Auth.can_manage?(user_id, org_id),
-         {:ok, :valid} <- Validator.validate(notification),
-         {:ok, n} <- create_notification(org_id, notification) do
+         {:ok, :valid} <- Validator.validate(notification, user_id),
+         {:ok, n} <- create_notification(org_id, user_id, notification) do
       Serialization.serialize(n)
     else
       {:error, :permission_denied} ->
@@ -34,12 +34,13 @@ defmodule Notifications.Api.PublicApi.Create do
     end
   end
 
-  def create_notification(org_id, notification) do
+  def create_notification(org_id, creator_id, notification) do
     Repo.transaction(fn ->
       n =
         Models.Notification.new(
           org_id,
           notification.metadata.name,
+          creator_id,
           Notifications.Util.Transforms.encode_spec(notification.spec)
         )
 

notifications/lib/notifications/api/public_api/update.ex

@@ -12,9 +12,9 @@ defmodule Notifications.Api.PublicApi.Update do
     Logger.info("#{inspect(org_id)} #{inspect(user_id)} #{name_or_id}")
 
     with {:ok, :authorized} <- Auth.can_manage?(user_id, org_id),
-         {:ok, :valid} <- Validator.validate(req.notification),
+         {:ok, :valid} <- Validator.validate(req.notification, user_id),
          {:ok, n} <- Model.find_by_id_or_name(org_id, name_or_id),
-         {:ok, n} <- update_notification(n, req.notification) do
+         {:ok, n} <- update_notification(n, user_id, req.notification) do
       Serialization.serialize(n)
     else
       {:error, :permission_denied} ->
@@ -44,11 +44,12 @@ defmodule Notifications.Api.PublicApi.Update do
   # notification: existing row from database
   # apiresource: new notification from the API call that needs to be applied
   #
-  def update_notification(notification, apiresource) do
+  def update_notification(notification, creator_id, apiresource) do
     Repo.transaction(fn ->
       changes =
         Model.changeset(notification, %{
           name: apiresource.metadata.name,
+          creator_id: creator_id,
           spec: Notifications.Util.Transforms.encode_spec(apiresource.spec)
         })
 

notifications/lib/notifications/auth.ex

@@ -12,8 +12,14 @@ defmodule Notifications.Auth do
     authorize(user_id, org_id, "organization.notifications.manage")
   end
 
-  defp authorize(user_id, org_id, permission) do
-    req = Request.new(user_id: user_id, org_id: org_id)
+  def can_view_project?(user_id, org_id, project_id) do
+    authorize(user_id, org_id, project_id, "project.view")
+  end
+
+  defp authorize(user_id, org_id, permission), do: authorize(user_id, org_id, "", permission)
+
+  defp authorize(user_id, org_id, project_id, permission) do
+    req = Request.new(user_id: user_id, org_id: org_id, project_id: project_id)
     endpoint = Application.fetch_env!(:notifications, :rbac_endpoint)
     {:ok, channel} = GRPC.Stub.connect(endpoint)
 

notifications/lib/notifications/models/notification.ex

@@ -12,16 +12,18 @@ defmodule Notifications.Models.Notification do
     has_many(:rules, Notifications.Models.Rule, on_delete: :delete_all)
 
     field(:org_id, :binary_id)
+    field(:creator_id, :binary_id)
     field(:name, :string)
     field(:spec, :map)
 
     timestamps()
   end
 
-  def new(org_id, name, spec) do
+  def new(org_id, name, creator_id, spec) do
     %__MODULE__{}
     |> changeset(%{
       org_id: org_id,
+      creator_id: creator_id,
       name: name,
       spec: spec
     })
@@ -78,8 +80,8 @@ defmodule Notifications.Models.Notification do
 
   def changeset(notification, params \\ %{}) do
     notification
-    |> cast(params, [:org_id, :name, :spec])
-    |> validate_required([:org_id, :name, :spec])
+    |> cast(params, [:org_id, :creator_id, :name, :spec])
+    |> validate_required([:org_id, :creator_id, :name, :spec])
     |> valid_name_format(params)
     |> unique_constraint(
       :unique_names,

notifications/lib/notifications/util/validator.ex

@@ -1,16 +1,24 @@
 defmodule Notifications.Util.Validator do
   alias Notifications.Models.Pattern
 
-  def validate(notification) do
+  def validate(notification, user_id) do
     with {:ok, :valid} <- validate_regex_patterns(notification),
          {:ok, :valid} <- validate_result_patterns(notification),
-         {:ok, :valid} <- validate_notification_targets(notification) do
+         {:ok, :valid} <- validate_notification_targets(notification),
+         {:ok, :valid} <- validate_user_id(user_id) do
       {:ok, :valid}
     else
       e -> e
     end
   end
 
+  defp validate_user_id(user_id) do
+    case Ecto.UUID.cast(user_id) do
+      {:ok, _} -> {:ok, :valid}
+      :error -> {:error, :invalid_argument, "Invalid user_id: expected a valid UUID"}
+    end
+  end
+
   def validate_notification_targets(notification = %InternalApi.Notifications.Notification{}),
     do: validate_notification_targets_(notification.rules)
 

notifications/lib/notifications/workers/coordinator.ex

@@ -50,7 +50,9 @@ defmodule Notifications.Workers.Coordinator do
           organization: organization
         }
 
-        rules |> Enum.each(fn rule -> process(request_id, rule, data) end)
+        rules
+        |> Enum.filter(&authorized?(&1.notification.creator_id, &1.org_id, project.metadata.id))
+        |> Enum.each(fn rule -> process(request_id, rule, data) end)
 
         Logger.info("#{request_id} #{event.pipeline_id}")
       end)
@@ -87,6 +89,15 @@ defmodule Notifications.Workers.Coordinator do
       Logger.info("#{request_id} [done]")
     end
 
+    defp authorized?(_creator_id = nil, _org_id, _project_id), do: true
+
+    defp authorized?(creator_id, org_id, project_id) do
+      case Notifications.Auth.can_view_project?(creator_id, org_id, project_id) do
+        {:ok, :authorized} -> true
+        _ -> false
+      end
+    end
+
     defp map_result_to_string(enum), do: enum |> Atom.to_string() |> String.downcase()
   end
 end

notifications/lib/notifications/workers/coordinator/api.ex

@@ -4,7 +4,6 @@ defmodule Notifications.Workers.Coordinator.Api do
     alias InternalApi.RepoProxy.DescribeRequest
 
     req = DescribeRequest.new(hook_id: hook_id)
-
     endpoint = Application.fetch_env!(:notifications, :repo_proxy_endpoint)
     {:ok, channel} = GRPC.Stub.connect(endpoint)
 
@@ -39,7 +38,6 @@ defmodule Notifications.Workers.Coordinator.Api do
     alias InternalApi.Plumber.PipelineService.Stub
 
     req = InternalApi.Plumber.DescribeRequest.new(ppl_id: pipeline_id, detailed: true)
-
     endpoint = Application.get_env(:notifications, :pipeline_endpoint)
     {:ok, channel} = GRPC.Stub.connect(endpoint)
 

notifications/lib/notifications/workers/coordinator/filter.ex

@@ -10,6 +10,7 @@ defmodule Notifications.Workers.Coordinator.Filter do
     |> with_pattern([branch, pr_branch], "branch")
     |> with_pattern(pipeline, "pipeline")
     |> with_pattern(result, "result")
+    |> preload(:notification)
     |> Repo.all()
   end
 

notifications/lib/public_api/semaphore/notifications.v1alpha.pb.ex

@@ -34,15 +34,17 @@ defmodule Semaphore.Notifications.V1alpha.Notification.Metadata do
           name: String.t(),
           id: String.t(),
           create_time: integer,
-          update_time: integer
+          update_time: integer,
+          creator_id: String.t()
         }
 
-  defstruct [:name, :id, :create_time, :update_time]
+  defstruct [:name, :id, :create_time, :update_time, :creator_id]
 
   field(:name, 1, type: :string)
   field(:id, 2, type: :string)
   field(:create_time, 3, type: :int64)
   field(:update_time, 4, type: :int64)
+  field(:creator_id, 5, type: :string)
 end
 
 defmodule Semaphore.Notifications.V1alpha.Notification.Spec.Rule.Filter do

notifications/mix.exs

@@ -37,6 +37,7 @@ defmodule Notifications.Mixfile do
       {:ecto_sql, "~> 3.0"},
       {:postgrex, ">= 0.0.0"},
       {:grpc_mock, github: "renderedtext/grpc-mock", only: [:dev, :test]},
+      {:mock, "~> 0.3.0", only: [:dev, :test]},
       {:paginator, "~> 1.0"},
       {:tackle, github: "renderedtext/ex-tackle"},
       {:util, github: "renderedtext/elixir-util", branch: "rw/string_enums"},

notifications/mix.lock

@@ -29,8 +29,10 @@
   "jumper": {:hex, :jumper, "1.0.1", "3c00542ef1a83532b72269fab9f0f0c82bf23a35e27d278bfd9ed0865cecabff", [:mix], [], "hexpm", "318c59078ac220e966d27af3646026db9b5a5e6703cb2aa3e26bcfaba65b7433"},
   "junit_formatter": {:hex, :junit_formatter, "3.3.1", "c729befb848f1b9571f317d2fefa648e9d4869befc4b2980daca7c1edc468e40", [:mix], [], "hexpm", "761fc5be4b4c15d8ba91a6dafde0b2c2ae6db9da7b8832a55b5a1deb524da72b"},
   "lager": {:hex, :lager, "3.9.2", "4cab289120eb24964e3886bd22323cb5fefe4510c076992a23ad18cf85413d8c", [:rebar3], [{:goldrush, "0.1.9", [hex: :goldrush, repo: "hexpm", optional: false]}], "hexpm", "7f904d9e87a8cb7e66156ed31768d1c8e26eba1d54f4bc85b1aa4ac1f6340c28"},
+  "meck": {:hex, :meck, "0.9.2", "85ccbab053f1db86c7ca240e9fc718170ee5bda03810a6292b5306bf31bae5f5", [:rebar3], [], "hexpm", "81344f561357dc40a8344afa53767c32669153355b626ea9fcbc8da6b3045826"},
   "metrics": {:hex, :metrics, "1.0.1", "25f094dea2cda98213cecc3aeff09e940299d950904393b2a29d191c346a8486", [:rebar3], [], "hexpm", "69b09adddc4f74a40716ae54d140f93beb0fb8978d8636eaded0c31b6f099f16"},
   "mimerl": {:hex, :mimerl, "1.2.0", "67e2d3f571088d5cfd3e550c383094b47159f3eee8ffa08e64106cdf5e981be3", [:rebar3], [], "hexpm", "f278585650aa581986264638ebf698f8bb19df297f66ad91b18910dfc6e19323"},
+  "mock": {:hex, :mock, "0.3.9", "10e44ad1f5962480c5c9b9fa779c6c63de9bd31997c8e04a853ec990a9d841af", [:mix], [{:meck, "~> 0.9.2", [hex: :meck, repo: "hexpm", optional: false]}], "hexpm", "9e1b244c4ca2551bb17bb8415eed89e40ee1308e0fbaed0a4fdfe3ec8a4adbd3"},
   "paginator": {:hex, :paginator, "1.1.0", "e502a91b08d14a6c8319f86eae027e67b0c699ffca4d76afbcb43eea5f32cd53", [:mix], [{:ecto, "~> 3.0", [hex: :ecto, repo: "hexpm", optional: false]}, {:ecto_sql, "~> 3.0", [hex: :ecto_sql, repo: "hexpm", optional: false]}, {:plug_crypto, "~> 1.2.0", [hex: :plug_crypto, repo: "hexpm", optional: false]}, {:postgrex, "~> 0.13", [hex: :postgrex, repo: "hexpm", optional: true]}], "hexpm", "b484f533b4a157ff60df7c988c9ef04e9c64e1f96e74a48a700717517a1ba5d9"},
   "parse_trans": {:hex, :parse_trans, "3.3.1", "16328ab840cc09919bd10dab29e431da3af9e9e7e7e6f0089dd5a2d2820011d8", [:rebar3], [], "hexpm", "07cd9577885f56362d414e8c4c4e6bdf10d43a8767abb92d24cbe8b24c54888b"},
   "plug_crypto": {:hex, :plug_crypto, "1.2.2", "05654514ac717ff3a1843204b424477d9e60c143406aa94daf2274fdd280794d", [:mix], [], "hexpm", "87631c7ad914a5a445f0a3809f99b079113ae4ed4b867348dd9eec288cecb6db"},

notifications/priv/repo/migrations/20250804163141_add_creator_id_to_notifications.exs

@@ -0,0 +1,9 @@
+defmodule Notifications.Repo.Migrations.AddCreatorIdToNotifications do
+  use Ecto.Migration
+
+  def change do
+    alter table(:notifications) do
+      add :creator_id, :binary_id, null: true
+    end
+  end
+end