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