defmodule Ash.Error.Forbidden.Policy do @moduledoc "Raised when policy authorization for an action fails" require Logger use Ash.Error.Exception alias Ash.Policy.Policy def_ash_error( [ scenarios: [], facts: %{}, filter: nil, policy_breakdown?: false, must_pass_strict_check?: false, policies: [], resource: nil, action: nil ], class: :forbidden ) def exception(opts) do exception = super(Keyword.put(opts, :policy_breakdown?, Ash.Policy.Info.show_policy_breakdowns?())) case Ash.Policy.Info.log_policy_breakdowns() do nil -> :ok level -> Logger.log(level, report(exception, help_text?: false)) end exception end @help_text """ A check status of `?` implies that the solver did not need to determine that check. Some checks may look like they failed when in reality there was simply no need to check them. Look for policies with `✘` and `✓` in check statuses. A check with a `⬇` means that it didn't determine if the policy was authorized or forbidden, and so moved on to the next check. `🌟` and `⛔` mean that the check was responsible for producing an authorized or forbidden (respectively) status. If no check results in a status (they all have `⬇`) then the policy is assumed to have failed. In some cases, however, the policy may have just been ignored, as described above. """ @doc """ Print a report of an authorization failure Options: - `:help_text?`: Defaults to true. Displays help text at the top of the policy breakdown. """ def report(error, opts \\ []) do error |> get_errors() |> case do [] -> "No policy errors" errors -> error_lines = errors |> Enum.map(fn %{ facts: facts, filter: filter, policies: policies, must_pass_strict_check?: must_pass_strict_check? } -> must_pass_strict_check? = if must_pass_strict_check? do """ Scenario must pass strict check only, meaning `runtime` policies cannot be checked. This requirement is generally used for filtering on related resources, when we can't fetch those related resources to run `runtime` policies. For this reason, you generally want your primary read actions on your resources to have standard policies which can be checked statically (like `actor_attribute_equals`) in addition to filter policies, like `expr(foo == :bar)`. """ else "" end policy_breakdown_title = if Keyword.get(opts, :help_text?, true) do ["Policy Breakdown", @help_text] else "Policy Breakdown" end policy_explanation = policies |> Enum.filter(&relevant?(&1, facts)) |> Enum.map(&explain_policy(&1, facts)) |> Enum.intersperse("\n") |> title(policy_breakdown_title, false) filter = if filter do title( "Did not match filter expression #{inspect(filter)}", "Generated Filter" ) else "" end [must_pass_strict_check?, filter, policy_explanation] |> Enum.filter(& &1) |> Enum.intersperse("\n\n") end) |> Enum.intersperse("\n\n") [title_line(error), "\n", error_lines] |> IO.iodata_to_binary() |> String.trim() end end defp title_line(error) do cond do error.resource && error.action -> "#{inspect(error.resource)}.#{action_name(error.action)}" error.resource -> "#{inspect(error.resource)}" error.action -> "#{action_name(error.action)}" end end defp action_name(%{name: name}), do: name defp action_name(name), do: name defp relevant?(policy, facts) do Enum.all?(policy.condition, fn condition -> Policy.fetch_fact(facts, condition) == {:ok, true} end) end defp title(other, title, semicolon \\ true) defp title([], _, _), do: [] defp title(other, title, true), do: [title, ":\n", other] defp title(other, title, false), do: [title, "\n", other] defp explain_policy(policy, facts) do bypass = if policy.bypass? do "Bypass: " else "" end {description, state} = describe_checks(policy.policies, facts) tag = case state do :unknown -> "⛔" :authorized -> "🌟" :forbidden -> "⛔" end title(Enum.map(description, &[" ", &1]), [ " ", bypass, policy.description || "Policy", " | ", tag ]) end defp describe_checks(checks, facts) do {description, state} = Enum.reduce(checks, {[], :unknown}, fn check, {descriptions, state} -> new_state = case state do :unknown -> new_state( check.type, Policy.fetch_fact(facts, check.check) ) other -> other end tag = case {state, new_state} do {:unknown, :authorized} -> "🌟" {:unknown, :forbidden} -> "⛔" {:unknown, :unknown} -> "⬇" _ -> "" end {[describe_check(check, Policy.fetch_fact(facts, check.check), tag) | descriptions], new_state} end) {Enum.intersperse(Enum.reverse(description), "\n"), state} end defp describe_check(check, fact_result, tag) do fact_result = case fact_result do {:ok, true} -> "✓" {:ok, false} -> "✘" :error -> "?" end [ check_type(check), ": ", check.check_module.describe(check.check_opts), " | ", fact_result, " | ", tag ] end defp check_type(%{type: :authorize_if}), do: "authorize if" defp check_type(%{type: :forbid_if}), do: "forbid if" defp check_type(%{type: :authorize_unless}), do: "authorize unless" defp check_type(%{type: :forbid_unless}), do: "forbid unless" defp new_state(:authorize_if, {:ok, true}), do: :authorized defp new_state(:forbid_if, {:ok, true}), do: :forbidden defp new_state(:authorize_unless, {:ok, false}), do: :authorized defp new_state(:forbid_unless, {:ok, false}), do: :forbidden defp new_state(_, _), do: :unknown defp get_errors(%Ash.Error.Forbidden{errors: errors}) do Enum.flat_map(errors || [], fn error -> get_errors(error) end) end defp get_errors(%__MODULE__{} = error) do [error] end defp get_errors(_), do: [] defimpl Ash.ErrorKind do def id(_), do: Ecto.UUID.generate() def message(error) do if error.policy_breakdown? do "forbidden:\n\n#{Ash.Error.Forbidden.Policy.report(error, help_text?: false)}" else "forbidden" end end def code(_), do: "Forbidden" end end