Packages

phoenix_kit

1.7.113
1.7.208 1.7.207 1.7.206 1.7.205 1.7.204 1.7.203 1.7.202 1.7.201 1.7.200 1.7.199 1.7.198 1.7.197 1.7.196 1.7.194 1.7.193 1.7.192 1.7.191 1.7.190 1.7.189 1.7.187 1.7.186 1.7.185 1.7.184 1.7.183 1.7.182 1.7.181 1.7.180 1.7.179 1.7.178 1.7.177 1.7.176 1.7.175 1.7.174 1.7.173 1.7.172 1.7.171 1.7.170 1.7.169 1.7.168 1.7.167 1.7.166 1.7.165 1.7.164 1.7.162 1.7.161 1.7.160 1.7.159 1.7.157 1.7.156 1.7.155 1.7.154 1.7.153 1.7.152 1.7.151 1.7.150 1.7.149 1.7.146 1.7.145 1.7.144 1.7.143 1.7.138 1.7.133 1.7.132 1.7.131 1.7.130 1.7.128 1.7.126 1.7.125 1.7.121 1.7.120 1.7.119 1.7.118 1.7.117 1.7.116 1.7.115 1.7.114 1.7.113 1.7.112 1.7.111 1.7.110 1.7.109 1.7.108 1.7.107 1.7.106 1.7.105 1.7.104 1.7.103 1.7.102 1.7.101 1.7.100 1.7.99 1.7.98 1.7.97 1.7.96 1.7.95 1.7.94 1.7.93 1.7.92 1.7.91 1.7.90 1.7.89 1.7.88 1.7.87 1.7.86 1.7.85 1.7.84 1.7.83 1.7.82 1.7.81 1.7.80 1.7.79 1.7.78 1.7.77 1.7.76 1.7.75 1.7.74 1.7.71 1.7.70 1.7.69 1.7.66 1.7.65 1.7.64 1.7.63 1.7.62 1.7.61 1.7.59 1.7.58 1.7.57 1.7.56 1.7.55 1.7.54 1.7.53 1.7.52 1.7.51 1.7.49 1.7.44 1.7.43 1.7.42 1.7.41 1.7.39 1.7.38 1.7.37 1.7.36 1.7.34 1.7.33 1.7.31 1.7.30 1.7.29 1.7.28 1.7.27 1.7.26 1.7.25 1.7.24 1.7.23 1.7.22 1.7.21 1.7.20 1.7.19 1.7.18 1.7.17 1.7.16 1.7.15 1.7.14 1.7.13 1.7.12 1.7.11 1.7.10 1.7.9 1.7.8 1.7.7 1.7.6 1.7.5 1.7.4 1.7.3 1.7.2 1.7.1 1.7.0 1.6.20 1.6.19 1.6.18 1.6.17 1.6.16 1.6.15 1.6.14 1.6.13 1.6.12 1.6.11 1.6.10 1.6.9 1.6.8 1.6.7 1.6.6 1.6.5 1.6.4 1.6.3 1.5.2 1.5.1 1.5.0 1.4.9 1.4.8 1.4.7 1.4.6 1.4.5 1.4.4 1.4.3 1.4.2 1.4.1 1.4.0 1.3.2 1.3.1 1.3.0 1.2.10 1.2.9 1.2.8 1.2.7 1.2.5 1.2.4 1.2.2 1.2.1 1.2.0 1.1.0 1.0.0

A foundation for building Elixir Phoenix apps — SaaS, social networks, ERP systems, marketplaces, and more

Current section

Files

Jump to
phoenix_kit lib phoenix_kit users permissions.ex
Raw

lib/phoenix_kit/users/permissions.ex

defmodule PhoenixKit.Users.Permissions do
@moduledoc """
Context for module-level permissions in PhoenixKit.
Controls which roles can access which admin sections and feature modules.
Uses an allowlist model: row present = granted, absent = denied.
Owner role always has full access (enforced in code, no DB rows needed).
## Permission Keys
Core sections: dashboard, users, media, settings, modules
Feature modules: billing, shop, emails, entities, tickets, posts, comments,
ai, publishing, referrals, sitemap, seo, maintenance, storage,
languages, connections, legal, db, jobs
## Constants & Metadata
Permissions.all_module_keys() # 25 built-in + any custom keys
Permissions.core_section_keys() # 5 core keys
Permissions.feature_module_keys() # 20 feature keys
Permissions.enabled_module_keys() # Core + enabled features + custom keys
Permissions.valid_module_key?("ai") # true
Permissions.feature_enabled?("ai") # true/false based on module status
Permissions.module_label("shop") # "E-Commerce"
Permissions.module_icon("shop") # "hero-shopping-cart"
Permissions.module_description("shop") # "Product catalog, orders, ..."
## Query API
Permissions.get_permissions_for_user(user) # User's keys via roles
Permissions.get_permissions_for_role(role_uuid) # Keys for a role
Permissions.role_has_permission?(role_uuid, "billing") # Single check
Permissions.get_permissions_matrix() # All roles → MapSet
Permissions.roles_with_permission("billing") # Role UUIDs with key
Permissions.users_with_permission("billing") # User UUIDs with key
Permissions.count_permissions_for_role(role_uuid) # Efficient count
Permissions.diff_permissions(role_a, role_b) # Compare two roles
## Mutation API
Permissions.grant_permission(role_uuid, "billing", granted_by_uuid)
Permissions.revoke_permission(role_uuid, "billing")
Permissions.set_permissions(role_uuid, ["dashboard", "users"], granted_by_uuid)
Permissions.grant_all_permissions(role_uuid, granted_by_uuid)
Permissions.revoke_all_permissions(role_uuid)
Permissions.copy_permissions(source_role_uuid, target_role_uuid, granted_by_uuid)
## Custom Keys API
Parent apps can register custom permission keys for custom admin tabs:
Permissions.register_custom_key("analytics", label: "Analytics", icon: "hero-chart-bar")
Permissions.unregister_custom_key("analytics")
Permissions.custom_keys() # List of registered custom key strings
Permissions.custom_view_permissions() # %{ViewModule => "key"} mapping
Custom keys are always treated as "enabled" (no module toggle) and appear
in the permission matrix UI under a "Custom" group.
## Edit Protection
Permissions.can_edit_role_permissions?(scope, role) :: :ok | {:error, String.t()}
Enforces: users cannot edit their own role, only Owner can edit Admin,
system roles cannot have `is_system_role` changed.
"""
import Ecto.Query, warn: false
require Logger
alias PhoenixKit.Admin.Events
alias PhoenixKit.ModuleRegistry
alias PhoenixKit.RepoHelper
alias PhoenixKit.Settings
alias PhoenixKit.Users.Auth.Scope
alias PhoenixKit.Users.Auth.User
alias PhoenixKit.Users.Role
alias PhoenixKit.Users.RoleAssignment
alias PhoenixKit.Users.RolePermission
alias PhoenixKit.Users.Roles
alias PhoenixKit.Users.ScopeNotifier
alias PhoenixKit.Utils.Date, as: UtilsDate
@core_section_keys ~w(dashboard users media settings modules)
# Persistent term keys for runtime-registered custom permission keys
@custom_keys_pterm {PhoenixKit, :custom_permission_keys}
@custom_views_pterm {PhoenixKit, :custom_view_permissions}
@valid_key_pattern ~r/^[a-z][a-z0-9_]*$/
@max_key_length 50
@max_custom_keys 50
@max_label_length 100
@max_icon_length 60
@max_description_length 255
# Feature enabled checks are now resolved at runtime via ModuleRegistry.feature_enabled_checks/0
# --- Custom Permission Keys ---
@doc """
Registers a custom permission key with metadata.
Custom keys extend the built-in 25 permission keys, allowing parent apps
to define new permission scopes for custom admin tabs. Custom keys are
always treated as "enabled" (no module toggle) and appear in the
permission matrix UI under "Custom".
Raises `ArgumentError` if the key collides with a built-in key or has
an invalid format. Logs a warning on duplicate override.
## Options
- `:label` - Human-readable label (default: capitalized key)
- `:icon` - Heroicon name (default: `"hero-squares-2x2"`)
- `:description` - Short description (default: `""`)
## Examples
Permissions.register_custom_key("analytics", label: "Analytics", icon: "hero-chart-bar")
"""
@spec register_custom_key(String.t(), keyword()) :: :ok
def register_custom_key(key, opts \\ []) when is_binary(key) do
if key in @core_section_keys or key in ModuleRegistry.all_feature_keys() do
raise ArgumentError,
"Cannot register custom permission key #{inspect(key)}: conflicts with built-in key"
end
unless Regex.match?(@valid_key_pattern, key) do
raise ArgumentError,
"Invalid permission key #{inspect(key)}: must match ~r/^[a-z][a-z0-9_]*$/"
end
if String.length(key) > @max_key_length do
raise ArgumentError,
"Permission key #{inspect(key)} exceeds max length of #{@max_key_length}"
end
meta = %{
label:
opts
|> Keyword.get(:label)
|> coerce_string(String.capitalize(key))
|> String.slice(0, @max_label_length),
icon:
opts
|> Keyword.get(:icon)
|> coerce_string("hero-squares-2x2")
|> String.slice(0, @max_icon_length),
description:
opts
|> Keyword.get(:description)
|> coerce_string("")
|> String.slice(0, @max_description_length)
}
# Note: persistent_term has no CAS, so concurrent register_custom_key calls
# could theoretically exceed the limit by 1-2 keys. This is acceptable since
# registration only happens at app startup, not at runtime.
# Re-registration of existing keys is always allowed (override).
current = custom_keys_map()
if not Map.has_key?(current, key) and map_size(current) >= @max_custom_keys do
raise ArgumentError,
"Cannot register more than #{@max_custom_keys} custom permission keys"
end
if Map.has_key?(current, key) do
Logger.warning(
"[Permissions] Custom permission key #{inspect(key)} re-registered, overriding previous metadata"
)
end
:persistent_term.put(@custom_keys_pterm, Map.put(current, key, meta))
# Auto-grant custom keys to Admin role so they have access by default.
# Uses a settings flag to avoid re-granting after Owner explicitly revokes.
auto_grant_to_admin_roles(key)
:ok
end
@doc """
Unregisters a custom permission key. Stale DB rows are harmless.
"""
@spec unregister_custom_key(String.t()) :: :ok
def unregister_custom_key(key) when is_binary(key) do
current = custom_keys_map()
:persistent_term.put(@custom_keys_pterm, Map.delete(current, key))
# Clean up any view → permission mappings that reference this key
views = :persistent_term.get(@custom_views_pterm, %{})
cleaned =
views
|> Enum.reject(fn {_mod, perm} -> perm == key end)
|> Map.new()
if map_size(cleaned) != map_size(views) do
:persistent_term.put(@custom_views_pterm, cleaned)
end
# Clear auto-grant flag so re-registering the key will auto-grant again
clear_auto_grant_flag(key)
:ok
end
@doc """
Returns the map of registered custom permission keys and their metadata.
"""
@spec custom_keys_map() :: %{String.t() => map()}
def custom_keys_map do
:persistent_term.get(@custom_keys_pterm, %{})
end
@doc """
Returns the list of custom permission key strings.
"""
@spec custom_keys() :: [String.t()]
def custom_keys do
custom_keys_map() |> Map.keys() |> Enum.sort()
end
@doc """
Clears all custom permission keys. For test isolation.
"""
@spec clear_custom_keys() :: :ok
def clear_custom_keys do
:persistent_term.put(@custom_keys_pterm, %{})
:persistent_term.put(@custom_views_pterm, %{})
:ok
end
@doc """
Caches a LiveView module → permission key mapping for custom admin tabs.
Used by the auth system to enforce permissions on custom admin LiveViews
without reading Application config on every mount.
"""
@spec cache_custom_view_permission(module(), String.t()) :: :ok
def cache_custom_view_permission(view_module, permission_key)
when is_atom(view_module) and is_binary(permission_key) do
current = :persistent_term.get(@custom_views_pterm, %{})
case Map.get(current, view_module) do
nil ->
:ok
^permission_key ->
:ok
old_key ->
Logger.warning(
"[Permissions] View #{inspect(view_module)} permission changed from #{inspect(old_key)} to #{inspect(permission_key)}"
)
end
:persistent_term.put(@custom_views_pterm, Map.put(current, view_module, permission_key))
:ok
end
@doc """
Returns the cached custom view → permission mapping.
"""
@spec custom_view_permissions() :: %{module() => String.t()}
def custom_view_permissions do
:persistent_term.get(@custom_views_pterm, %{})
end
# --- Constants ---
@doc "Returns all built-in and custom permission keys as a list. See `enabled_module_keys/0` for filtered MapSet variant."
@spec all_module_keys() :: [String.t()]
def all_module_keys, do: @core_section_keys ++ feature_module_keys() ++ custom_keys()
@doc "Returns the 5 core section keys."
@spec core_section_keys() :: [String.t()]
def core_section_keys, do: @core_section_keys
@doc "Returns the feature module keys from the registry."
@spec feature_module_keys() :: [String.t()]
def feature_module_keys, do: ModuleRegistry.all_feature_keys()
@doc """
Returns module keys that are currently enabled (core sections + enabled feature modules + custom keys)
as a `MapSet` for efficient membership checks. Core sections and custom keys are always included.
Feature modules are included only if their module reports enabled status.
Returns `MapSet.t()` unlike `all_module_keys/0` which returns a list — callers use this
primarily for `MapSet.member?/2` and `MapSet.intersection/2` checks.
"""
@spec enabled_module_keys() :: MapSet.t()
def enabled_module_keys do
enabled_features =
feature_module_keys()
|> Enum.filter(&do_feature_enabled?/1)
MapSet.new(@core_section_keys ++ enabled_features ++ custom_keys())
end
@doc "Checks whether `key` is a known permission key (built-in or custom)."
@spec valid_module_key?(String.t()) :: boolean()
def valid_module_key?(key) when is_binary(key) do
key in @core_section_keys or
key in ModuleRegistry.all_feature_keys() or
Map.has_key?(custom_keys_map(), key)
end
def valid_module_key?(_), do: false
@doc """
Checks whether a feature module is currently enabled.
Core section keys always return `true`. Feature module keys return the
result of calling the module's `enabled?/0` (or equivalent) function.
Custom permission keys are always enabled (no module toggle).
Returns `false` for unknown keys.
"""
@spec feature_enabled?(String.t()) :: boolean()
def feature_enabled?(key) when key in @core_section_keys, do: true
def feature_enabled?(key) when is_binary(key) do
case Map.get(ModuleRegistry.feature_enabled_checks(), key) do
{mod, fun} ->
Code.ensure_loaded?(mod) && apply(mod, fun, [])
nil ->
# Custom keys are always "enabled" (no module toggle)
Map.has_key?(custom_keys_map(), key)
end
rescue
_ -> false
end
# Core section metadata (always present, not from registry)
@core_labels %{
"dashboard" => "Dashboard",
"users" => "Users",
"media" => "Media",
"settings" => "Settings",
"modules" => "Modules",
# `db` was extracted into `phoenix_kit_db` but core still
# references the key (e.g. `auth.ex` `/admin/db` route).
# `phoenix_kit_db` registers `"db" => "DB"` via its
# `permission_metadata/0`, but only when the module is loaded
# in the parent app. Core's own test environment doesn't load
# external modules, so without this fallback `module_label("db")`
# produces `String.capitalize("db")` = `"Db"`. Pin the canonical
# label here so the display is correct regardless of whether
# the external module is installed.
"db" => "DB"
}
@core_icons %{
"dashboard" => "hero-home",
"users" => "hero-users",
"media" => "hero-photo",
"settings" => "hero-cog-6-tooth",
"modules" => "hero-squares-2x2",
# Mirrors `phoenix_kit_db`'s registered icon. See `@core_labels`
# above for the rationale.
"db" => "hero-server-stack"
}
@core_descriptions %{
"dashboard" => "Overview statistics, charts, and system health",
"users" => "User accounts, roles, and access management",
"media" => "File uploads, image processing, and storage buckets",
"settings" => "General, organization, and user preference settings",
"modules" => "Enable, disable, and configure feature modules",
# Mirrors `phoenix_kit_db`'s registered description. See
# `@core_labels` above for the rationale.
"db" => "Database explorer and schema inspection"
}
@doc "Returns a human-readable label for a module key."
@spec module_label(String.t()) :: String.t()
def module_label(key) do
Map.get_lazy(@core_labels, key, fn ->
case Map.get(ModuleRegistry.permission_labels(), key) do
nil -> custom_key_metadata(key)[:label] || String.capitalize(key)
label -> label
end
end)
end
@doc "Returns a Heroicon name for a module key."
@spec module_icon(String.t()) :: String.t()
def module_icon(key) do
Map.get_lazy(@core_icons, key, fn ->
case Map.get(ModuleRegistry.permission_icons(), key) do
nil -> custom_key_metadata(key)[:icon] || "hero-squares-2x2"
icon -> icon
end
end)
end
@doc "Returns a short description for a module key."
@spec module_description(String.t()) :: String.t()
def module_description(key) do
Map.get_lazy(@core_descriptions, key, fn ->
case Map.get(ModuleRegistry.permission_descriptions(), key) do
nil -> custom_key_metadata(key)[:description] || ""
desc -> desc
end
end)
end
# --- Query API ---
@doc """
Returns the list of module_keys the given user has access to.
Joins through role_assignments → role_permissions.
"""
@spec get_permissions_for_user(User.t() | nil) :: [String.t()]
def get_permissions_for_user(nil), do: []
def get_permissions_for_user(%User{uuid: nil}), do: []
def get_permissions_for_user(%User{uuid: user_uuid}) when not is_nil(user_uuid) do
repo = RepoHelper.repo()
from(rp in RolePermission,
join: ra in RoleAssignment,
on: ra.role_uuid == rp.role_uuid,
where: ra.user_uuid == ^user_uuid,
select: rp.module_key,
distinct: true
)
|> repo.all()
rescue
e ->
if table_missing_error?(e) do
Logger.error(
"PhoenixKit: phoenix_kit_role_permissions table not found. " <>
"Run `mix phoenix_kit.update` to apply V53 migration."
)
else
Logger.warning("Permissions.get_permissions_for_user failed: #{inspect(e)}")
end
[]
end
@doc """
Checks if a specific role has a specific permission.
"""
@spec role_has_permission?(String.t(), String.t()) :: boolean()
def role_has_permission?(role_uuid, module_key) do
repo = RepoHelper.repo()
role_uuid = resolve_role_uuid(role_uuid)
from(rp in RolePermission,
where: rp.role_uuid == ^role_uuid and rp.module_key == ^module_key,
select: true
)
|> repo.exists?()
rescue
e ->
Logger.warning("Permissions.role_has_permission? failed: #{inspect(e)}")
false
end
@doc """
Returns the list of module_keys granted to a specific role.
"""
@spec get_permissions_for_role(String.t()) :: [String.t()]
def get_permissions_for_role(role_uuid) do
repo = RepoHelper.repo()
role_uuid = resolve_role_uuid(role_uuid)
from(rp in RolePermission,
where: rp.role_uuid == ^role_uuid,
select: rp.module_key,
order_by: [asc: rp.module_key]
)
|> repo.all()
rescue
e ->
Logger.warning("Permissions.get_permissions_for_role failed: #{inspect(e)}")
[]
end
@doc """
Returns a matrix of role_uuid → MapSet of granted keys for all roles.
"""
@spec get_permissions_matrix() :: %{String.t() => MapSet.t()}
def get_permissions_matrix do
repo = RepoHelper.repo()
from(rp in RolePermission,
select: {rp.role_uuid, rp.module_key}
)
|> repo.all()
|> Enum.group_by(&elem(&1, 0), &elem(&1, 1))
|> Map.new(fn {role_uuid, keys} -> {role_uuid, MapSet.new(keys)} end)
rescue
e ->
Logger.warning("Permissions.get_permissions_matrix failed: #{inspect(e)}")
%{}
end
@doc """
Returns a list of role_ids that have been granted the given module_key.
"""
@spec roles_with_permission(String.t()) :: [String.t()]
def roles_with_permission(module_key) do
repo = RepoHelper.repo()
from(rp in RolePermission,
where: rp.module_key == ^module_key,
select: rp.role_uuid,
order_by: [asc: rp.role_uuid]
)
|> repo.all()
rescue
e ->
Logger.warning("Permissions.roles_with_permission failed: #{inspect(e)}")
[]
end
@doc """
Returns a list of user_ids that have access to the given module_key
(through any of their assigned roles).
"""
@spec users_with_permission(String.t()) :: [String.t()]
def users_with_permission(module_key) do
repo = RepoHelper.repo()
from(rp in RolePermission,
join: ra in RoleAssignment,
on: ra.role_uuid == rp.role_uuid,
where: rp.module_key == ^module_key,
select: ra.user_uuid,
distinct: true,
order_by: [asc: ra.user_uuid]
)
|> repo.all()
rescue
e ->
Logger.warning("Permissions.users_with_permission failed: #{inspect(e)}")
[]
end
@doc """
Returns the number of permission keys granted to a role.
More efficient than `length(get_permissions_for_role(role_uuid))`.
"""
@spec count_permissions_for_role(integer() | String.t()) :: non_neg_integer()
def count_permissions_for_role(role_uuid) do
repo = RepoHelper.repo()
role_uuid = resolve_role_uuid(role_uuid)
from(rp in RolePermission,
where: rp.role_uuid == ^role_uuid,
select: count()
)
|> repo.one()
rescue
e ->
Logger.warning("Permissions.count_permissions_for_role failed: #{inspect(e)}")
0
end
@doc """
Compares permissions between two roles and returns a diff map.
Returns `%{only_a: MapSet.t(), only_b: MapSet.t(), common: MapSet.t()}`
where `only_a` are keys role_a has but role_b doesn't, `only_b` is the
inverse, and `common` are keys both roles share.
"""
@spec diff_permissions(integer() | String.t(), integer() | String.t()) :: %{
only_a: MapSet.t(),
only_b: MapSet.t(),
common: MapSet.t()
}
def diff_permissions(role_uuid_a, role_uuid_b) do
keys_a = get_permissions_for_role(role_uuid_a) |> MapSet.new()
keys_b = get_permissions_for_role(role_uuid_b) |> MapSet.new()
%{
only_a: MapSet.difference(keys_a, keys_b),
only_b: MapSet.difference(keys_b, keys_a),
common: MapSet.intersection(keys_a, keys_b)
}
end
# --- Mutation API ---
@doc """
Grants a single permission to a role. Uses upsert to be idempotent.
"""
@spec grant_permission(integer() | String.t(), String.t(), integer() | String.t() | nil) ::
{:ok, RolePermission.t()} | {:error, Ecto.Changeset.t()}
def grant_permission(role_uuid, module_key, granted_by_uuid \\ nil) do
repo = RepoHelper.repo()
role_uuid = resolve_role_uuid(role_uuid)
if is_nil(role_uuid) do
{:error, :role_not_found}
else
grant_permission_insert(repo, role_uuid, module_key, granted_by_uuid)
end
end
defp grant_permission_insert(repo, role_uuid, module_key, granted_by_uuid) do
granted_by_uuid = resolve_user_uuid(granted_by_uuid)
%RolePermission{}
|> RolePermission.changeset(%{
role_uuid: role_uuid,
module_key: module_key,
granted_by_uuid: granted_by_uuid
})
|> repo.insert(
on_conflict: :nothing,
conflict_target: [:role_uuid, :module_key]
)
|> tap(fn
{:ok, %{uuid: uuid}} when not is_nil(uuid) ->
Events.broadcast_permission_granted(role_uuid, module_key)
notify_affected_users(role_uuid)
_ ->
:ok
end)
end
@doc """
Revokes a single permission from a role.
"""
@spec revoke_permission(integer() | String.t(), String.t()) :: :ok | {:error, :not_found}
def revoke_permission(role_uuid, module_key) do
repo = RepoHelper.repo()
role_uuid = resolve_role_uuid(role_uuid)
from(rp in RolePermission,
where: rp.role_uuid == ^role_uuid and rp.module_key == ^module_key
)
|> repo.delete_all()
|> case do
{0, _} ->
{:error, :not_found}
{_, _} ->
Events.broadcast_permission_revoked(role_uuid, module_key)
notify_affected_users(role_uuid)
:ok
end
end
@doc """
Syncs permissions for a role: grants missing keys, revokes extras.
Runs in a transaction.
"""
@spec set_permissions(integer() | String.t(), [String.t()], integer() | String.t() | nil) ::
:ok | {:error, term()}
def set_permissions(role_uuid, desired_keys, granted_by_uuid \\ nil) do
repo = RepoHelper.repo()
valid_keys = MapSet.new(all_module_keys())
repo.transaction(fn ->
role_uuid = resolve_role_uuid(role_uuid)
# Lock existing permission rows FOR UPDATE to prevent concurrent set_permissions
# from reading the same state and computing conflicting diffs.
current_keys =
from(rp in RolePermission,
where: rp.role_uuid == ^role_uuid,
select: rp.module_key,
lock: "FOR UPDATE"
)
|> repo.all()
|> MapSet.new()
# Filter out any invalid keys before processing
desired_set = desired_keys |> MapSet.new() |> MapSet.intersection(valid_keys)
# Keys to add
to_add = MapSet.difference(desired_set, current_keys)
# Keys to remove
to_remove = MapSet.difference(current_keys, desired_set)
# Bulk insert new permissions
if MapSet.size(to_add) > 0 do
now = UtilsDate.utc_now()
granted_by_uuid = resolve_user_uuid(granted_by_uuid)
entries =
Enum.map(to_add, fn key ->
%{
uuid: UUIDv7.generate(),
role_uuid: role_uuid,
module_key: key,
granted_by_uuid: granted_by_uuid,
inserted_at: now
}
end)
repo.insert_all(RolePermission, entries, on_conflict: :nothing)
end
# Bulk delete removed permissions
if MapSet.size(to_remove) > 0 do
remove_list = MapSet.to_list(to_remove)
from(rp in RolePermission,
where: rp.role_uuid == ^role_uuid and rp.module_key in ^remove_list
)
|> repo.delete_all()
end
MapSet.to_list(desired_set)
end)
|> case do
{:ok, filtered_keys} ->
Events.broadcast_permissions_synced(role_uuid, filtered_keys)
notify_affected_users(role_uuid)
:ok
{:error, reason} ->
{:error, reason}
end
end
@doc """
Grants all permission keys (built-in + custom) to a role.
"""
@spec grant_all_permissions(integer() | String.t(), integer() | String.t() | nil) ::
:ok | {:error, term()}
def grant_all_permissions(role_uuid, granted_by_uuid \\ nil) do
set_permissions(role_uuid, all_module_keys(), granted_by_uuid)
end
@doc """
Revokes all permissions from a role.
"""
@spec revoke_all_permissions(integer() | String.t()) :: :ok | {:error, term()}
def revoke_all_permissions(role_uuid) do
repo = RepoHelper.repo()
role_uuid = resolve_role_uuid(role_uuid)
from(rp in RolePermission, where: rp.role_uuid == ^role_uuid)
|> repo.delete_all()
Events.broadcast_permissions_synced(role_uuid, [])
notify_affected_users(role_uuid)
:ok
rescue
e ->
require Logger
Logger.warning("[PhoenixKit.Permissions] revoke_all_permissions failed: #{inspect(e)}")
{:error, e}
end
@doc """
Copies all permissions from one role to another.
The target role will end up with the exact same set of permissions as the
source role. Existing permissions on the target that don't exist on the
source will be revoked.
"""
@spec copy_permissions(
integer() | String.t(),
integer() | String.t(),
integer() | String.t() | nil
) :: :ok | {:error, term()}
def copy_permissions(source_role_uuid, target_role_uuid, granted_by_uuid \\ nil) do
source_keys = get_permissions_for_role(source_role_uuid)
set_permissions(target_role_uuid, source_keys, granted_by_uuid)
end
# --- Access Control ---
@doc """
Checks if the given scope can edit the target role's permissions.
Returns `:ok` if allowed, or `{:error, reason}` if not.
Rules:
- Owner role cannot be edited (always has full access)
- Users cannot edit their own role (prevents self-lockout)
- Only Owner can edit Admin role (prevents privilege escalation)
"""
@spec can_edit_role_permissions?(Scope.t() | nil, Role.t()) :: :ok | {:error, atom()}
def can_edit_role_permissions?(nil, _role), do: {:error, :not_authenticated}
def can_edit_role_permissions?(scope, role) do
if Scope.authenticated?(scope) do
can_edit_role_permissions_check(scope, role)
else
{:error, :not_authenticated}
end
end
defp can_edit_role_permissions_check(scope, role) do
user_roles = Scope.user_roles(scope)
cond do
role.name == "Owner" ->
{:error, :owner_immutable}
role.name in user_roles and not Scope.system_role?(scope) ->
{:error, :self_role}
role.name == "Admin" and not Scope.owner?(scope) ->
{:error, :admin_owner_only}
true ->
:ok
end
end
# --- Helpers ---
# Returns metadata for a custom permission key, or nil if not found.
defp custom_key_metadata(key) do
Map.get(custom_keys_map(), key)
end
defp do_feature_enabled?(key) do
case Map.get(ModuleRegistry.feature_enabled_checks(), key) do
{mod, fun} ->
Code.ensure_loaded?(mod) && apply(mod, fun, [])
nil ->
false
end
rescue
_ -> false
end
# Detect Postgrex "relation does not exist" errors (table missing)
defp table_missing_error?(%{postgres: %{code: :undefined_table}}), do: true
defp table_missing_error?(%Postgrex.Error{postgres: %{code: :undefined_table}}), do: true
defp table_missing_error?(%{message: msg}) when is_binary(msg) do
String.contains?(msg, "does not exist")
end
defp table_missing_error?(_), do: false
# Resolves an integer role_id to its UUID for changeset/insert_all use
defp resolve_role_uuid(nil), do: nil
defp resolve_role_uuid(role_uuid) when is_binary(role_uuid), do: role_uuid
defp resolve_user_uuid(nil), do: nil
defp resolve_user_uuid(user_uuid) when is_binary(user_uuid), do: user_uuid
# Notify all users with the affected role to refresh their scope
defp notify_affected_users(role_uuid) do
repo = RepoHelper.repo()
role_uuid = resolve_role_uuid(role_uuid)
user_uuids =
from(ra in RoleAssignment,
where: ra.role_uuid == ^role_uuid,
select: ra.user_uuid
)
|> repo.all()
Enum.each(user_uuids, &ScopeNotifier.broadcast_roles_updated/1)
rescue
e ->
Logger.warning("Permissions.notify_affected_users failed: #{inspect(e)}")
:ok
end
# Clears the auto-grant settings flag for a custom key so that
# re-registering it will trigger a fresh auto-grant to Admin.
defp clear_auto_grant_flag(key) do
Settings.update_setting("auto_granted_perm:#{key}", nil)
rescue
_ -> :ok
end
@doc """
Auto-grants a permission key to the Admin system role.
Stores a flag in phoenix_kit_settings so that if Owner later revokes
the key, it won't be re-granted on next application restart.
"""
@spec auto_grant_to_admin_roles(String.t()) :: :ok
def auto_grant_to_admin_roles(key) do
flag_key = "auto_granted_perm:#{key}"
# If already auto-granted before, respect any manual changes
if Settings.get_setting(flag_key) == "true" do
:ok
else
case Roles.get_role_by_name(Role.system_roles().admin) do
%{uuid: admin_uuid} when not is_nil(admin_uuid) ->
case grant_permission(admin_uuid, key, nil) do
{:ok, _} ->
Settings.update_setting(flag_key, "true")
{:error, _} ->
Logger.warning(
"[Permissions] grant_permission failed for Admin role on key #{inspect(key)}, will retry next boot"
)
end
:ok
_ ->
# Admin role not found (pre-V53 or missing), skip
:ok
end
end
rescue
error ->
msg = Exception.message(error)
# Silently skip if the table doesn't exist yet (expected on fresh installs
# where the app starts before migrations have created the table)
unless String.contains?(msg, "undefined_table") do
Logger.warning("[Permissions] Failed to auto-grant #{inspect(key)} to Admin role: #{msg}")
end
:ok
end
# Coerces a value to a string, returning the default for nil.
# Handles atoms, integers, and other types gracefully via to_string/1.
defp coerce_string(nil, default), do: default
defp coerce_string(value, _default) when is_binary(value), do: value
defp coerce_string(value, _default), do: to_string(value)
end