Implement M6 push notification policy guards and replay tests

PROGRESS_MARMOT.md

@@ -48,9 +48,9 @@ Spec source: `~/marmot/README.md` + MIP-00..05.
 
 ## M6 — optional MIP-05 (push notifications)
 
-- [ ] Accept/store notification coordination events required by enabled profile
+- [x] Accept/store notification coordination events required by enabled profile
-- [ ] Add policy/rate-limit controls for push-related event traffic
+- [x] Add policy/rate-limit controls for push-related event traffic
-- [ ] Add abuse and replay protection tests for notification trigger paths
+- [x] Add abuse and replay protection tests for notification trigger paths
 
 ## M7 — hardening + operations
 

config/config.exs

@@ -26,13 +26,21 @@ config :parrhesia,
     marmot_media_max_url_bytes: 2048,
     marmot_media_allowed_mime_prefixes: [],
     marmot_media_reject_mip04_v1: true,
+    marmot_push_server_pubkeys: [],
+    marmot_push_max_relay_tags: 16,
+    marmot_push_max_payload_bytes: 65_536,
+    marmot_push_max_trigger_age_seconds: 120,
+    marmot_push_require_expiration: true,
+    marmot_push_max_expiration_window_seconds: 120,
+    marmot_push_max_server_recipients: 1,
     management_auth_required: true
   ],
   features: [
     nip_45_count: true,
     nip_50_search: true,
     nip_77_negentropy: true,
-    nip_ee_mls: false
+    nip_ee_mls: false,
+    marmot_push_notifications: false
   ],
   storage: [
     events: Parrhesia.Storage.Adapters.Postgres.Events,

lib/parrhesia/policy/event_policy.ex

@@ -19,6 +19,12 @@ defmodule Parrhesia.Policy.EventPolicy do
           | :media_metadata_invalid_mime
           | :media_metadata_mime_not_allowed
           | :media_metadata_unsupported_version
+          | :push_notification_relay_tags_exceeded
+          | :push_notification_payload_too_large
+          | :push_notification_replay_window_exceeded
+          | :push_notification_missing_expiration
+          | :push_notification_expiration_too_far
+          | :push_notification_server_recipients_exceeded
           | :protected_event_requires_auth
           | :protected_event_pubkey_mismatch
           | :pow_below_minimum
@@ -51,6 +57,7 @@ defmodule Parrhesia.Policy.EventPolicy do
       fn -> enforce_pow(event) end,
       fn -> enforce_protected_event(event, authenticated_pubkeys) end,
       fn -> enforce_media_metadata_policy(event) end,
+      fn -> enforce_push_notification_policy(event) end,
       fn -> enforce_mls_feature_flag(event) end
     ]
 
@@ -101,6 +108,24 @@ defmodule Parrhesia.Policy.EventPolicy do
   def error_message(:media_metadata_unsupported_version),
     do: "blocked: media metadata version is not supported"
 
+  def error_message(:push_notification_relay_tags_exceeded),
+    do: "rate-limited: push relay list contains too many relay tags"
+
+  def error_message(:push_notification_payload_too_large),
+    do: "rate-limited: push notification payload exceeds configured size limit"
+
+  def error_message(:push_notification_replay_window_exceeded),
+    do: "restricted: push notification trigger is outside replay window"
+
+  def error_message(:push_notification_missing_expiration),
+    do: "invalid: push notification trigger requires an expiration tag"
+
+  def error_message(:push_notification_expiration_too_far),
+    do: "invalid: push notification expiration exceeds configured window"
+
+  def error_message(:push_notification_server_recipients_exceeded),
+    do: "rate-limited: push notification trigger targets too many notification servers"
+
   def error_message(:protected_event_requires_auth),
     do: "auth-required: protected events require authenticated pubkey"
 
@@ -401,6 +426,145 @@ defmodule Parrhesia.Policy.EventPolicy do
 
   defp valid_mime_type?(_mime_type), do: false
 
+  defp enforce_push_notification_policy(event) do
+    if config_bool([:features, :marmot_push_notifications], false) do
+      case Map.get(event, "kind") do
+        10_050 -> validate_push_relay_list_event(event)
+        1059 -> maybe_validate_push_trigger_event(event)
+        _other -> :ok
+      end
+    else
+      :ok
+    end
+  end
+
+  defp validate_push_relay_list_event(event) do
+    relay_tags =
+      event
+      |> Map.get("tags", [])
+      |> Enum.filter(fn
+        ["relay", _url | _rest] -> true
+        _tag -> false
+      end)
+
+    max_relay_tags = config_int([:policies, :marmot_push_max_relay_tags], 16)
+
+    if max_relay_tags > 0 and length(relay_tags) > max_relay_tags do
+      {:error, :push_notification_relay_tags_exceeded}
+    else
+      :ok
+    end
+  end
+
+  defp maybe_validate_push_trigger_event(event) do
+    push_server_pubkeys = config_list([:policies, :marmot_push_server_pubkeys], [])
+
+    if targets_push_server?(event, push_server_pubkeys) do
+      with :ok <- validate_push_payload_size(event),
+           :ok <- validate_push_replay_window(event),
+           :ok <- validate_push_expiration(event) do
+        validate_push_server_recipient_count(event, push_server_pubkeys)
+      end
+    else
+      :ok
+    end
+  end
+
+  defp targets_push_server?(event, push_server_pubkeys) do
+    event
+    |> recipient_pubkeys()
+    |> Enum.any?(&(&1 in push_server_pubkeys))
+  end
+
+  defp validate_push_payload_size(event) do
+    max_payload_bytes = config_int([:policies, :marmot_push_max_payload_bytes], 65_536)
+    content = Map.get(event, "content", "")
+
+    if is_binary(content) and (max_payload_bytes <= 0 or byte_size(content) <= max_payload_bytes) do
+      :ok
+    else
+      {:error, :push_notification_payload_too_large}
+    end
+  end
+
+  defp validate_push_replay_window(event) do
+    max_age_seconds = config_int([:policies, :marmot_push_max_trigger_age_seconds], 120)
+    created_at = Map.get(event, "created_at")
+    now = System.system_time(:second)
+
+    if max_age_seconds <= 0 or
+         (is_integer(created_at) and created_at >= 0 and now - created_at <= max_age_seconds) do
+      :ok
+    else
+      {:error, :push_notification_replay_window_exceeded}
+    end
+  end
+
+  defp validate_push_expiration(event) do
+    require_expiration? = config_bool([:policies, :marmot_push_require_expiration], true)
+
+    case expiration_tag_value(event) do
+      nil ->
+        if require_expiration?, do: {:error, :push_notification_missing_expiration}, else: :ok
+
+      expiration when is_integer(expiration) ->
+        max_expiration_window =
+          config_int([:policies, :marmot_push_max_expiration_window_seconds], 120)
+
+        created_at = Map.get(event, "created_at")
+
+        if max_expiration_window <= 0 or
+             (is_integer(created_at) and expiration >= created_at and
+                expiration - created_at <= max_expiration_window) do
+          :ok
+        else
+          {:error, :push_notification_expiration_too_far}
+        end
+    end
+  end
+
+  defp validate_push_server_recipient_count(event, push_server_pubkeys) do
+    max_server_recipients = config_int([:policies, :marmot_push_max_server_recipients], 1)
+
+    target_count =
+      event
+      |> recipient_pubkeys()
+      |> Enum.count(&(&1 in push_server_pubkeys))
+
+    if max_server_recipients > 0 and target_count > max_server_recipients do
+      {:error, :push_notification_server_recipients_exceeded}
+    else
+      :ok
+    end
+  end
+
+  defp recipient_pubkeys(event) do
+    event
+    |> Map.get("tags", [])
+    |> Enum.reduce([], fn
+      ["p", recipient | _rest], acc -> [recipient | acc]
+      _tag, acc -> acc
+    end)
+  end
+
+  defp expiration_tag_value(event) do
+    event
+    |> Map.get("tags", [])
+    |> Enum.find_value(fn
+      ["expiration", unix_seconds | _rest] -> parse_unix_seconds(unix_seconds)
+      _tag -> nil
+    end)
+  end
+
+  defp parse_unix_seconds(unix_seconds) when is_binary(unix_seconds) do
+    case Integer.parse(unix_seconds) do
+      {parsed, ""} when parsed >= 0 -> parsed
+      _other -> nil
+    end
+  end
+
+  defp parse_unix_seconds(_unix_seconds), do: nil
+
   defp reject_if_pubkey_banned(event) do
     with pubkey when is_binary(pubkey) <- Map.get(event, "pubkey"),
          {:ok, true} <- Storage.moderation().pubkey_banned?(%{}, pubkey) do

lib/parrhesia/web/connection.ex

@@ -438,6 +438,12 @@ defmodule Parrhesia.Web.Connection do
               :media_metadata_invalid_mime,
               :media_metadata_mime_not_allowed,
               :media_metadata_unsupported_version,
+              :push_notification_relay_tags_exceeded,
+              :push_notification_payload_too_large,
+              :push_notification_replay_window_exceeded,
+              :push_notification_missing_expiration,
+              :push_notification_expiration_too_far,
+              :push_notification_server_recipients_exceeded,
               :mls_disabled
             ],
        do: EventPolicy.error_message(reason)

test/parrhesia/config_test.exs

@@ -10,8 +10,10 @@ defmodule Parrhesia.ConfigTest do
     assert Parrhesia.Config.get([:policies, :auth_required_for_writes]) == false
     assert Parrhesia.Config.get([:policies, :marmot_media_max_imeta_tags_per_event]) == 8
     assert Parrhesia.Config.get([:policies, :marmot_media_reject_mip04_v1]) == true
+    assert Parrhesia.Config.get([:policies, :marmot_push_max_trigger_age_seconds]) == 120
     assert Parrhesia.Config.get([:features, :nip_50_search]) == true
     assert Parrhesia.Config.get([:features, :nip_ee_mls]) == false
+    assert Parrhesia.Config.get([:features, :marmot_push_notifications]) == false
   end
 
   test "returns default for unknown keys" do

test/parrhesia/policy/event_policy_test.exs

@@ -223,6 +223,207 @@ defmodule Parrhesia.Policy.EventPolicyTest do
              )
   end
 
+  test "accepts push coordination events when push feature is enabled" do
+    server_pubkey = String.duplicate("f", 64)
+
+    Application.put_env(
+      :parrhesia,
+      :features,
+      nip_ee_mls: false,
+      marmot_push_notifications: true
+    )
+
+    Application.put_env(
+      :parrhesia,
+      :policies,
+      marmot_push_server_pubkeys: [server_pubkey],
+      marmot_push_max_relay_tags: 16,
+      marmot_push_max_payload_bytes: 65_536,
+      marmot_push_max_trigger_age_seconds: 300,
+      marmot_push_require_expiration: true,
+      marmot_push_max_expiration_window_seconds: 120,
+      marmot_push_max_server_recipients: 1
+    )
+
+    relay_list_event = %{
+      "kind" => 10_050,
+      "tags" => [["relay", "wss://notify.example"], ["relay", "wss://notify2.example"]],
+      "pubkey" => String.duplicate("d", 64),
+      "id" => "",
+      "created_at" => System.system_time(:second),
+      "content" => ""
+    }
+
+    now = System.system_time(:second)
+
+    trigger_event = %{
+      "kind" => 1059,
+      "tags" => [["p", server_pubkey], ["expiration", Integer.to_string(now + 60)]],
+      "pubkey" => String.duplicate("d", 64),
+      "id" => "",
+      "created_at" => now,
+      "content" => "encrypted-push"
+    }
+
+    assert :ok =
+             EventPolicy.authorize_write(
+               relay_list_event,
+               MapSet.new([String.duplicate("d", 64)])
+             )
+
+    assert :ok =
+             EventPolicy.authorize_write(trigger_event, MapSet.new([String.duplicate("d", 64)]))
+  end
+
+  test "enforces push policy limits for relay-list and trigger payloads" do
+    server_pubkey = String.duplicate("e", 64)
+
+    Application.put_env(
+      :parrhesia,
+      :features,
+      nip_ee_mls: false,
+      marmot_push_notifications: true
+    )
+
+    Application.put_env(
+      :parrhesia,
+      :policies,
+      marmot_push_server_pubkeys: [server_pubkey],
+      marmot_push_max_relay_tags: 1,
+      marmot_push_max_payload_bytes: 8,
+      marmot_push_max_trigger_age_seconds: 300,
+      marmot_push_require_expiration: true,
+      marmot_push_max_expiration_window_seconds: 120,
+      marmot_push_max_server_recipients: 1
+    )
+
+    relay_list_event = %{
+      "kind" => 10_050,
+      "tags" => [["relay", "wss://notify.example"], ["relay", "wss://notify2.example"]],
+      "pubkey" => String.duplicate("d", 64),
+      "id" => "",
+      "created_at" => System.system_time(:second),
+      "content" => ""
+    }
+
+    now = System.system_time(:second)
+
+    oversized_trigger = %{
+      "kind" => 1059,
+      "tags" => [["p", server_pubkey], ["expiration", Integer.to_string(now + 60)]],
+      "pubkey" => String.duplicate("d", 64),
+      "id" => "",
+      "created_at" => now,
+      "content" => "encrypted-push-too-large"
+    }
+
+    assert {:error, :push_notification_relay_tags_exceeded} =
+             EventPolicy.authorize_write(
+               relay_list_event,
+               MapSet.new([String.duplicate("d", 64)])
+             )
+
+    assert {:error, :push_notification_payload_too_large} =
+             EventPolicy.authorize_write(
+               oversized_trigger,
+               MapSet.new([String.duplicate("d", 64)])
+             )
+  end
+
+  test "enforces push replay and expiration protection" do
+    server_pubkey = String.duplicate("c", 64)
+    now = System.system_time(:second)
+
+    Application.put_env(
+      :parrhesia,
+      :features,
+      nip_ee_mls: false,
+      marmot_push_notifications: true
+    )
+
+    Application.put_env(
+      :parrhesia,
+      :policies,
+      marmot_push_server_pubkeys: [server_pubkey],
+      marmot_push_max_relay_tags: 16,
+      marmot_push_max_payload_bytes: 65_536,
+      marmot_push_max_trigger_age_seconds: 5,
+      marmot_push_require_expiration: true,
+      marmot_push_max_expiration_window_seconds: 30,
+      marmot_push_max_server_recipients: 1
+    )
+
+    stale_trigger = %{
+      "kind" => 1059,
+      "tags" => [["p", server_pubkey], ["expiration", Integer.to_string(now - 50)]],
+      "pubkey" => String.duplicate("d", 64),
+      "id" => "",
+      "created_at" => now - 20,
+      "content" => "encrypted"
+    }
+
+    missing_expiration = %{
+      "kind" => 1059,
+      "tags" => [["p", server_pubkey]],
+      "pubkey" => String.duplicate("d", 64),
+      "id" => "",
+      "created_at" => now,
+      "content" => "encrypted"
+    }
+
+    far_expiration = %{
+      "kind" => 1059,
+      "tags" => [["p", server_pubkey], ["expiration", Integer.to_string(now + 120)]],
+      "pubkey" => String.duplicate("d", 64),
+      "id" => "",
+      "created_at" => now,
+      "content" => "encrypted"
+    }
+
+    multi_server_target = %{
+      "kind" => 1059,
+      "tags" => [
+        ["p", server_pubkey],
+        ["p", String.duplicate("b", 64)],
+        ["expiration", Integer.to_string(now + 20)]
+      ],
+      "pubkey" => String.duplicate("d", 64),
+      "id" => "",
+      "created_at" => now,
+      "content" => "encrypted"
+    }
+
+    Application.put_env(
+      :parrhesia,
+      :policies,
+      marmot_push_server_pubkeys: [server_pubkey, String.duplicate("b", 64)],
+      marmot_push_max_relay_tags: 16,
+      marmot_push_max_payload_bytes: 65_536,
+      marmot_push_max_trigger_age_seconds: 5,
+      marmot_push_require_expiration: true,
+      marmot_push_max_expiration_window_seconds: 30,
+      marmot_push_max_server_recipients: 1
+    )
+
+    assert {:error, :push_notification_replay_window_exceeded} =
+             EventPolicy.authorize_write(stale_trigger, MapSet.new([String.duplicate("d", 64)]))
+
+    assert {:error, :push_notification_missing_expiration} =
+             EventPolicy.authorize_write(
+               missing_expiration,
+               MapSet.new([String.duplicate("d", 64)])
+             )
+
+    assert {:error, :push_notification_expiration_too_far} =
+             EventPolicy.authorize_write(far_expiration, MapSet.new([String.duplicate("d", 64)]))
+
+    assert {:error, :push_notification_server_recipients_exceeded} =
+             EventPolicy.authorize_write(
+               multi_server_target,
+               MapSet.new([String.duplicate("d", 64)])
+             )
+  end
+
   test "rejects mls kinds when feature is disabled" do
     Application.put_env(:parrhesia, :features, nip_ee_mls: false)
 

test/parrhesia/web/conformance_test.exs

@@ -149,6 +149,82 @@ defmodule Parrhesia.Web.ConformanceTest do
     assert persisted_welcome["id"] == wrapped_welcome["id"]
   end
 
+  test "push coordination events are accepted and stored when feature is enabled" do
+    previous_features = Application.get_env(:parrhesia, :features, [])
+    previous_policies = Application.get_env(:parrhesia, :policies, [])
+
+    server_pubkey = String.duplicate("f", 64)
+
+    Application.put_env(
+      :parrhesia,
+      :features,
+      previous_features
+      |> Keyword.put(:marmot_push_notifications, true)
+      |> Keyword.put(:nip_ee_mls, false)
+    )
+
+    Application.put_env(
+      :parrhesia,
+      :policies,
+      previous_policies
+      |> Keyword.put(:marmot_push_server_pubkeys, [server_pubkey])
+      |> Keyword.put(:marmot_push_max_trigger_age_seconds, 300)
+      |> Keyword.put(:marmot_push_require_expiration, true)
+      |> Keyword.put(:marmot_push_max_expiration_window_seconds, 120)
+    )
+
+    on_exit(fn ->
+      Application.put_env(:parrhesia, :features, previous_features)
+      Application.put_env(:parrhesia, :policies, previous_policies)
+    end)
+
+    {:ok, state} = Connection.init(subscription_index: nil)
+
+    relay_list_event =
+      valid_event(%{
+        "kind" => 10_050,
+        "tags" => [["relay", "wss://notify.example"], ["relay", "wss://notify2.example"]],
+        "content" => ""
+      })
+
+    now = System.system_time(:second)
+
+    push_trigger =
+      valid_event(%{
+        "kind" => 1059,
+        "created_at" => now,
+        "tags" => [["p", server_pubkey], ["expiration", Integer.to_string(now + 60)]],
+        "content" => "encrypted-push"
+      })
+
+    assert {:push, {:text, relay_ok_frame}, ^state} =
+             Connection.handle_in(
+               {Jason.encode!(["EVENT", relay_list_event]), [opcode: :text]},
+               state
+             )
+
+    assert Jason.decode!(relay_ok_frame) == [
+             "OK",
+             relay_list_event["id"],
+             true,
+             "ok: event stored"
+           ]
+
+    assert {:push, {:text, trigger_ok_frame}, ^state} =
+             Connection.handle_in(
+               {Jason.encode!(["EVENT", push_trigger]), [opcode: :text]},
+               state
+             )
+
+    assert Jason.decode!(trigger_ok_frame) == ["OK", push_trigger["id"], true, "ok: event stored"]
+
+    assert {:ok, persisted_relay_list} = Storage.events().get_event(%{}, relay_list_event["id"])
+    assert persisted_relay_list["id"] == relay_list_event["id"]
+
+    assert {:ok, persisted_trigger} = Storage.events().get_event(%{}, push_trigger["id"])
+    assert persisted_trigger["id"] == push_trigger["id"]
+  end
+
   defp valid_event(overrides \\ %{}) do
     now = System.system_time(:second)
 

test/parrhesia/web/connection_test.exs

@@ -223,6 +223,117 @@ defmodule Parrhesia.Web.ConnectionTest do
            ]
   end
 
+  test "push trigger EVENT outside replay window is rejected" do
+    previous_features = Application.get_env(:parrhesia, :features, [])
+    previous_policies = Application.get_env(:parrhesia, :policies, [])
+
+    server_pubkey = String.duplicate("e", 64)
+
+    Application.put_env(
+      :parrhesia,
+      :features,
+      previous_features
+      |> Keyword.put(:marmot_push_notifications, true)
+      |> Keyword.put(:nip_ee_mls, false)
+    )
+
+    Application.put_env(
+      :parrhesia,
+      :policies,
+      previous_policies
+      |> Keyword.put(:marmot_push_server_pubkeys, [server_pubkey])
+      |> Keyword.put(:marmot_push_max_trigger_age_seconds, 5)
+      |> Keyword.put(:marmot_push_require_expiration, true)
+      |> Keyword.put(:marmot_push_max_expiration_window_seconds, 30)
+    )
+
+    on_exit(fn ->
+      Application.put_env(:parrhesia, :features, previous_features)
+      Application.put_env(:parrhesia, :policies, previous_policies)
+    end)
+
+    state = connection_state()
+    now = System.system_time(:second)
+
+    event =
+      valid_event()
+      |> Map.put("kind", 1059)
+      |> Map.put("created_at", now - 20)
+      |> Map.put("tags", [["p", server_pubkey], ["expiration", Integer.to_string(now - 5)]])
+      |> Map.put("content", "encrypted")
+      |> then(&Map.put(&1, "id", EventValidator.compute_id(&1)))
+
+    payload = Jason.encode!(["EVENT", event])
+
+    assert {:push, {:text, response}, ^state} =
+             Connection.handle_in({payload, [opcode: :text]}, state)
+
+    assert Jason.decode!(response) == [
+             "OK",
+             event["id"],
+             false,
+             "restricted: push notification trigger is outside replay window"
+           ]
+  end
+
+  test "duplicate push trigger EVENT is rejected" do
+    previous_features = Application.get_env(:parrhesia, :features, [])
+    previous_policies = Application.get_env(:parrhesia, :policies, [])
+
+    server_pubkey = String.duplicate("f", 64)
+
+    Application.put_env(
+      :parrhesia,
+      :features,
+      previous_features
+      |> Keyword.put(:marmot_push_notifications, true)
+      |> Keyword.put(:nip_ee_mls, false)
+    )
+
+    Application.put_env(
+      :parrhesia,
+      :policies,
+      previous_policies
+      |> Keyword.put(:marmot_push_server_pubkeys, [server_pubkey])
+      |> Keyword.put(:marmot_push_max_trigger_age_seconds, 300)
+      |> Keyword.put(:marmot_push_require_expiration, true)
+      |> Keyword.put(:marmot_push_max_expiration_window_seconds, 120)
+    )
+
+    on_exit(fn ->
+      Application.put_env(:parrhesia, :features, previous_features)
+      Application.put_env(:parrhesia, :policies, previous_policies)
+    end)
+
+    state = connection_state()
+    now = System.system_time(:second)
+
+    event =
+      valid_event()
+      |> Map.put("kind", 1059)
+      |> Map.put("created_at", now)
+      |> Map.put("tags", [["p", server_pubkey], ["expiration", Integer.to_string(now + 60)]])
+      |> Map.put("content", "encrypted")
+      |> then(&Map.put(&1, "id", EventValidator.compute_id(&1)))
+
+    payload = Jason.encode!(["EVENT", event])
+
+    assert {:push, {:text, first_response}, ^state} =
+             Connection.handle_in({payload, [opcode: :text]}, state)
+
+    assert Jason.decode!(first_response) == ["OK", event["id"], true, "ok: event stored"]
+
+    assert {:push, {:text, second_response}, ^state} =
+             Connection.handle_in({payload, [opcode: :text]}, state)
+
+    assert Jason.decode!(second_response) == [
+             "OK",
+             event["id"],
+             false,
+             "duplicate: event already stored"
+           ]
+  end
+
   test "NEG sessions open and close" do
     state = connection_state()