git.squeep.com Git - akkoma/blob - test/plugs/oauth_scopes_plug_test.exs

   1 # Pleroma: A lightweight social networking server
   2 # Copyright © 2017-2020 Pleroma Authors <https://pleroma.social/>
   3 # SPDX-License-Identifier: AGPL-3.0-only
   4
   5 defmodule Pleroma.Plugs.OAuthScopesPlugTest do
   6   use Pleroma.Web.ConnCase, async: true
   7
   8   alias Pleroma.Plugs.EnsurePublicOrAuthenticatedPlug
   9   alias Pleroma.Plugs.OAuthScopesPlug
  10   alias Pleroma.Repo
  11
  12   import Mock
  13   import Pleroma.Factory
  14
  15   setup_with_mocks([{EnsurePublicOrAuthenticatedPlug, [], [call: fn conn, _ -> conn end]}]) do
  16     :ok
  17   end
  18
  19   test "if `token.scopes` fulfills specified 'any of' conditions, " <>
  20          "proceeds with no op",
  21        %{conn: conn} do
  22     token = insert(:oauth_token, scopes: ["read", "write"]) |> Repo.preload(:user)
  23
  24     conn =
  25       conn
  26       |> assign(:user, token.user)
  27       |> assign(:token, token)
  28       |> OAuthScopesPlug.call(%{scopes: ["read"]})
  29
  30     refute conn.halted
  31     assert conn.assigns[:user]
  32   end
  33
  34   test "if `token.scopes` fulfills specified 'all of' conditions, " <>
  35          "proceeds with no op",
  36        %{conn: conn} do
  37     token = insert(:oauth_token, scopes: ["scope1", "scope2", "scope3"]) |> Repo.preload(:user)
  38
  39     conn =
  40       conn
  41       |> assign(:user, token.user)
  42       |> assign(:token, token)
  43       |> OAuthScopesPlug.call(%{scopes: ["scope2", "scope3"], op: :&})
  44
  45     refute conn.halted
  46     assert conn.assigns[:user]
  47   end
  48
  49   describe "with `fallback: :proceed_unauthenticated` option, " do
  50     test "if `token.scopes` doesn't fulfill specified conditions, " <>
  51            "clears :user and :token assigns and calls EnsurePublicOrAuthenticatedPlug",
  52          %{conn: conn} do
  53       user = insert(:user)
  54       token1 = insert(:oauth_token, scopes: ["read", "write"], user: user)
  55
  56       for token <- [token1, nil], op <- [:|, :&] do
  57         ret_conn =
  58           conn
  59           |> assign(:user, user)
  60           |> assign(:token, token)
  61           |> OAuthScopesPlug.call(%{
  62             scopes: ["follow"],
  63             op: op,
  64             fallback: :proceed_unauthenticated
  65           })
  66
  67         refute ret_conn.halted
  68         refute ret_conn.assigns[:user]
  69         refute ret_conn.assigns[:token]
  70
  71         assert called(EnsurePublicOrAuthenticatedPlug.call(ret_conn, :_))
  72       end
  73     end
  74
  75     test "with :skip_instance_privacy_check option, " <>
  76            "if `token.scopes` doesn't fulfill specified conditions, " <>
  77            "clears :user and :token assigns and does NOT call EnsurePublicOrAuthenticatedPlug",
  78          %{conn: conn} do
  79       user = insert(:user)
  80       token1 = insert(:oauth_token, scopes: ["read:statuses", "write"], user: user)
  81
  82       for token <- [token1, nil], op <- [:|, :&] do
  83         ret_conn =
  84           conn
  85           |> assign(:user, user)
  86           |> assign(:token, token)
  87           |> OAuthScopesPlug.call(%{
  88             scopes: ["read"],
  89             op: op,
  90             fallback: :proceed_unauthenticated,
  91             skip_instance_privacy_check: true
  92           })
  93
  94         refute ret_conn.halted
  95         refute ret_conn.assigns[:user]
  96         refute ret_conn.assigns[:token]
  97
  98         refute called(EnsurePublicOrAuthenticatedPlug.call(ret_conn, :_))
  99       end
 100     end
 101   end
 102
 103   describe "without :fallback option, " do
 104     test "if `token.scopes` does not fulfill specified 'any of' conditions, " <>
 105            "returns 403 and halts",
 106          %{conn: conn} do
 107       for token <- [insert(:oauth_token, scopes: ["read", "write"]), nil] do
 108         any_of_scopes = ["follow", "push"]
 109
 110         ret_conn =
 111           conn
 112           |> assign(:token, token)
 113           |> OAuthScopesPlug.call(%{scopes: any_of_scopes})
 114
 115         assert ret_conn.halted
 116         assert 403 == ret_conn.status
 117
 118         expected_error = "Insufficient permissions: #{Enum.join(any_of_scopes, " | ")}."
 119         assert Jason.encode!(%{error: expected_error}) == ret_conn.resp_body
 120       end
 121     end
 122
 123     test "if `token.scopes` does not fulfill specified 'all of' conditions, " <>
 124            "returns 403 and halts",
 125          %{conn: conn} do
 126       for token <- [insert(:oauth_token, scopes: ["read", "write"]), nil] do
 127         token_scopes = (token && token.scopes) || []
 128         all_of_scopes = ["write", "follow"]
 129
 130         conn =
 131           conn
 132           |> assign(:token, token)
 133           |> OAuthScopesPlug.call(%{scopes: all_of_scopes, op: :&})
 134
 135         assert conn.halted
 136         assert 403 == conn.status
 137
 138         expected_error =
 139           "Insufficient permissions: #{Enum.join(all_of_scopes -- token_scopes, " & ")}."
 140
 141         assert Jason.encode!(%{error: expected_error}) == conn.resp_body
 142       end
 143     end
 144   end
 145
 146   describe "with hierarchical scopes, " do
 147     test "if `token.scopes` fulfills specified 'any of' conditions, " <>
 148            "proceeds with no op",
 149          %{conn: conn} do
 150       token = insert(:oauth_token, scopes: ["read", "write"]) |> Repo.preload(:user)
 151
 152       conn =
 153         conn
 154         |> assign(:user, token.user)
 155         |> assign(:token, token)
 156         |> OAuthScopesPlug.call(%{scopes: ["read:something"]})
 157
 158       refute conn.halted
 159       assert conn.assigns[:user]
 160     end
 161
 162     test "if `token.scopes` fulfills specified 'all of' conditions, " <>
 163            "proceeds with no op",
 164          %{conn: conn} do
 165       token = insert(:oauth_token, scopes: ["scope1", "scope2", "scope3"]) |> Repo.preload(:user)
 166
 167       conn =
 168         conn
 169         |> assign(:user, token.user)
 170         |> assign(:token, token)
 171         |> OAuthScopesPlug.call(%{scopes: ["scope1:subscope", "scope2:subscope"], op: :&})
 172
 173       refute conn.halted
 174       assert conn.assigns[:user]
 175     end
 176   end
 177
 178   describe "filter_descendants/2" do
 179     test "filters scopes which directly match or are ancestors of supported scopes" do
 180       f = fn scopes, supported_scopes ->
 181         OAuthScopesPlug.filter_descendants(scopes, supported_scopes)
 182       end
 183
 184       assert f.(["read", "follow"], ["write", "read"]) == ["read"]
 185
 186       assert f.(["read", "write:something", "follow"], ["write", "read"]) ==
 187                ["read", "write:something"]
 188
 189       assert f.(["admin:read"], ["write", "read"]) == []
 190
 191       assert f.(["admin:read"], ["write", "admin"]) == ["admin:read"]
 192     end
 193   end
 194
 195   describe "transform_scopes/2" do
 196     setup do: clear_config([:auth, :enforce_oauth_admin_scope_usage])
 197
 198     setup do
 199       {:ok, %{f: &OAuthScopesPlug.transform_scopes/2}}
 200     end
 201
 202     test "with :admin option, prefixes all requested scopes with `admin:` " <>
 203            "and [optionally] keeps only prefixed scopes, " <>
 204            "depending on `[:auth, :enforce_oauth_admin_scope_usage]` setting",
 205          %{f: f} do
 206       Pleroma.Config.put([:auth, :enforce_oauth_admin_scope_usage], false)
 207
 208       assert f.(["read"], %{admin: true}) == ["admin:read", "read"]
 209
 210       assert f.(["read", "write"], %{admin: true}) == [
 211                "admin:read",
 212                "read",
 213                "admin:write",
 214                "write"
 215              ]
 216
 217       Pleroma.Config.put([:auth, :enforce_oauth_admin_scope_usage], true)
 218
 219       assert f.(["read:accounts"], %{admin: true}) == ["admin:read:accounts"]
 220
 221       assert f.(["read", "write:reports"], %{admin: true}) == [
 222                "admin:read",
 223                "admin:write:reports"
 224              ]
 225     end
 226
 227     test "with no supported options, returns unmodified scopes", %{f: f} do
 228       assert f.(["read"], %{}) == ["read"]
 229       assert f.(["read", "write"], %{}) == ["read", "write"]
 230     end
 231   end
 232 end