HaimKortovich
4519f797fd
All checks were successful
Build and Publish / build-release (push) Successful in 3m7s
82 lines
1.9 KiB
Elixir
82 lines
1.9 KiB
Elixir
defmodule CustomerServiceWeb.Plugs.AuthorizeRoles do
|
|
@moduledoc """
|
|
Authorize request based on Zitadel role permissions.
|
|
|
|
After token introspection, checks if the user holds any of the
|
|
`required_permissions` roles for the organization identified by
|
|
`X-Organization-Id` header.
|
|
|
|
The Zitadel roles claim structure is:
|
|
%{"urn:zitadel:iam:org:project:<project_id>:roles": {
|
|
"<role>": {
|
|
"<org_id>": "<org_domain>"
|
|
},
|
|
"<role>": {
|
|
"<org_id>": "<org_domain>"
|
|
}
|
|
}}
|
|
"""
|
|
|
|
@behaviour Plug
|
|
|
|
import Plug.Conn
|
|
|
|
@impl Plug
|
|
def init(opts),
|
|
do:
|
|
opts
|
|
|> Keyword.validate!([
|
|
:roles_claim
|
|
])
|
|
|
|
@impl Plug
|
|
def call(conn, opts) do
|
|
if authorized?(
|
|
conn,
|
|
Keyword.get(opts, :roles_claim),
|
|
Keyword.get(opts, :required_permissions)
|
|
) do
|
|
conn
|
|
else
|
|
conn
|
|
|> put_resp_content_type("application/json")
|
|
|> halt()
|
|
|> send_resp(
|
|
:forbidden,
|
|
Jason.encode_to_iodata!(%{error: "Forbidden", reason: "Missing required role"})
|
|
)
|
|
end
|
|
end
|
|
|
|
defp authorized?(conn, roles_claim, required_permissions) do
|
|
org_id = conn.private[CustomerServiceWeb.Plugs.ExtractOrganizationId]
|
|
|
|
with true <- org_id_given?(org_id),
|
|
roles_map <- get_roles_map(conn, roles_claim),
|
|
true <- has_any_role?(roles_map, org_id, required_permissions) do
|
|
true
|
|
else
|
|
_ -> false
|
|
end
|
|
end
|
|
|
|
defp org_id_given?(org_id), do: not is_nil(org_id)
|
|
|
|
defp get_roles_map(conn, roles_claim) do
|
|
case conn.private[Oidcc.Plug.IntrospectToken] do
|
|
%Oidcc.TokenIntrospection{extra: extra} ->
|
|
Map.get(extra, roles_claim, %{})
|
|
|
|
_ ->
|
|
%{}
|
|
end
|
|
end
|
|
|
|
defp has_any_role?(roles_map, org_id, required_permissions) do
|
|
Enum.any?(required_permissions, fn role ->
|
|
role_orgs = Map.get(roles_map, role, %{})
|
|
Map.has_key?(role_orgs, org_id)
|
|
end)
|
|
end
|
|
end
|