Graham McIntire
c3d952148e
- Remove unused Ecto.Query import from device_poller_worker_test - Update route paths from /orgs/:slug to /dashboard after navigation refactor - Fix HostParser tests to match actual hostname resolution behavior - Update UserAuth redirect assertions to expect /dashboard instead of /devices - Fix DashboardLiveTest missing organization context - Fix OrgLive.NewTest to verify organization creation correctly All 4850 tests now passing with zero failures and zero warnings.
1458 lines
46 KiB
Elixir
1458 lines
46 KiB
Elixir
defmodule ToweropsWeb.UserAuthTest do
|
|
use ToweropsWeb.ConnCase, async: true
|
|
use ExUnitProperties
|
|
|
|
import Towerops.AccountsFixtures
|
|
|
|
alias Phoenix.LiveView.Socket
|
|
alias Towerops.Accounts
|
|
alias Towerops.Accounts.Scope
|
|
alias ToweropsWeb.UserAuth
|
|
|
|
@remember_me_cookie "_towerops_web_user_remember_me"
|
|
@remember_me_cookie_max_age 60 * 60 * 24 * 14
|
|
|
|
setup %{conn: conn} do
|
|
conn =
|
|
conn
|
|
|> Map.replace!(:secret_key_base, ToweropsWeb.Endpoint.config(:secret_key_base))
|
|
|> init_test_session(%{})
|
|
|
|
%{user: %{user_fixture() | authenticated_at: DateTime.utc_now(:second)}, conn: conn}
|
|
end
|
|
|
|
describe "log_in_user/3" do
|
|
test "stores the user token in the session", %{conn: conn, user: user} do
|
|
conn = UserAuth.log_in_user(conn, user)
|
|
assert token = get_session(conn, :user_token)
|
|
assert redirected_to(conn) == ~p"/orgs"
|
|
assert Accounts.get_user_by_session_token(token)
|
|
end
|
|
|
|
test "clears everything previously stored in the session", %{conn: conn, user: user} do
|
|
conn = conn |> put_session(:to_be_removed, "value") |> UserAuth.log_in_user(user)
|
|
refute get_session(conn, :to_be_removed)
|
|
end
|
|
|
|
test "keeps session when re-authenticating", %{conn: conn, user: user} do
|
|
conn =
|
|
conn
|
|
|> assign(:current_scope, Scope.for_user(user))
|
|
|> put_session(:to_be_removed, "value")
|
|
|> UserAuth.log_in_user(user)
|
|
|
|
assert get_session(conn, :to_be_removed)
|
|
end
|
|
|
|
test "clears session when user does not match when re-authenticating", %{
|
|
conn: conn,
|
|
user: user
|
|
} do
|
|
other_user = user_fixture()
|
|
|
|
conn =
|
|
conn
|
|
|> assign(:current_scope, Scope.for_user(other_user))
|
|
|> put_session(:to_be_removed, "value")
|
|
|> UserAuth.log_in_user(user)
|
|
|
|
refute get_session(conn, :to_be_removed)
|
|
end
|
|
|
|
test "redirects to the configured path", %{conn: conn, user: user} do
|
|
conn = conn |> put_session(:user_return_to, "/hello") |> UserAuth.log_in_user(user)
|
|
assert redirected_to(conn) == "/hello"
|
|
end
|
|
|
|
test "writes a cookie if remember_me is configured", %{conn: conn, user: user} do
|
|
conn = conn |> fetch_cookies() |> UserAuth.log_in_user(user, %{"remember_me" => "true"})
|
|
assert get_session(conn, :user_token) == conn.cookies[@remember_me_cookie]
|
|
assert get_session(conn, :user_remember_me) == true
|
|
|
|
assert %{value: signed_token, max_age: max_age} = conn.resp_cookies[@remember_me_cookie]
|
|
assert signed_token != get_session(conn, :user_token)
|
|
assert max_age == @remember_me_cookie_max_age
|
|
end
|
|
|
|
test "writes a cookie if remember_me was set in previous session", %{conn: conn, user: user} do
|
|
conn = conn |> fetch_cookies() |> UserAuth.log_in_user(user, %{"remember_me" => "true"})
|
|
assert get_session(conn, :user_token) == conn.cookies[@remember_me_cookie]
|
|
assert get_session(conn, :user_remember_me) == true
|
|
|
|
conn =
|
|
conn
|
|
|> recycle()
|
|
|> Map.replace!(:secret_key_base, ToweropsWeb.Endpoint.config(:secret_key_base))
|
|
|> fetch_cookies()
|
|
|> init_test_session(%{user_remember_me: true})
|
|
|
|
# the conn is already logged in and has the remember_me cookie set,
|
|
# now we log in again and even without explicitly setting remember_me,
|
|
# the cookie should be set again
|
|
conn = UserAuth.log_in_user(conn, user, %{})
|
|
assert %{value: signed_token, max_age: max_age} = conn.resp_cookies[@remember_me_cookie]
|
|
assert signed_token != get_session(conn, :user_token)
|
|
assert max_age == @remember_me_cookie_max_age
|
|
assert get_session(conn, :user_remember_me) == true
|
|
end
|
|
end
|
|
|
|
describe "logout_user/1" do
|
|
test "erases session and cookies", %{conn: conn, user: user} do
|
|
user_token = Accounts.generate_user_session_token(user)
|
|
|
|
conn =
|
|
conn
|
|
|> put_session(:user_token, user_token)
|
|
|> put_req_cookie(@remember_me_cookie, user_token)
|
|
|> fetch_cookies()
|
|
|> UserAuth.log_out_user()
|
|
|
|
refute get_session(conn, :user_token)
|
|
refute conn.cookies[@remember_me_cookie]
|
|
assert %{max_age: 0} = conn.resp_cookies[@remember_me_cookie]
|
|
assert redirected_to(conn) == ~p"/users/log-in"
|
|
refute Accounts.get_user_by_session_token(user_token)
|
|
end
|
|
|
|
test "works even if user is already logged out", %{conn: conn} do
|
|
conn = conn |> fetch_cookies() |> UserAuth.log_out_user()
|
|
refute get_session(conn, :user_token)
|
|
assert %{max_age: 0} = conn.resp_cookies[@remember_me_cookie]
|
|
assert redirected_to(conn) == ~p"/users/log-in"
|
|
end
|
|
end
|
|
|
|
describe "fetch_current_scope_for_user/2" do
|
|
test "authenticates user from session", %{conn: conn, user: user} do
|
|
user_token = Accounts.generate_user_session_token(user)
|
|
|
|
conn =
|
|
conn |> put_session(:user_token, user_token) |> UserAuth.fetch_current_scope_for_user([])
|
|
|
|
assert conn.assigns.current_scope.user.id == user.id
|
|
assert conn.assigns.current_scope.user.authenticated_at == user.authenticated_at
|
|
assert get_session(conn, :user_token) == user_token
|
|
end
|
|
|
|
test "authenticates user from cookies", %{conn: conn, user: user} do
|
|
logged_in_conn =
|
|
conn |> fetch_cookies() |> UserAuth.log_in_user(user, %{"remember_me" => "true"})
|
|
|
|
user_token = logged_in_conn.cookies[@remember_me_cookie]
|
|
%{value: signed_token} = logged_in_conn.resp_cookies[@remember_me_cookie]
|
|
|
|
conn =
|
|
conn
|
|
|> put_req_cookie(@remember_me_cookie, signed_token)
|
|
|> UserAuth.fetch_current_scope_for_user([])
|
|
|
|
assert conn.assigns.current_scope.user.id == user.id
|
|
assert conn.assigns.current_scope.user.authenticated_at == user.authenticated_at
|
|
assert get_session(conn, :user_token) == user_token
|
|
assert get_session(conn, :user_remember_me)
|
|
end
|
|
|
|
test "does not authenticate if data is missing", %{conn: conn, user: user} do
|
|
_ = Accounts.generate_user_session_token(user)
|
|
conn = UserAuth.fetch_current_scope_for_user(conn, [])
|
|
refute get_session(conn, :user_token)
|
|
refute conn.assigns.current_scope
|
|
end
|
|
|
|
test "reissues a new token after a few days and refreshes cookie", %{conn: conn, user: user} do
|
|
logged_in_conn =
|
|
conn |> fetch_cookies() |> UserAuth.log_in_user(user, %{"remember_me" => "true"})
|
|
|
|
token = logged_in_conn.cookies[@remember_me_cookie]
|
|
%{value: signed_token} = logged_in_conn.resp_cookies[@remember_me_cookie]
|
|
|
|
offset_user_token(token, -10, :day)
|
|
{user, _} = Accounts.get_user_by_session_token(token)
|
|
|
|
conn =
|
|
conn
|
|
|> put_session(:user_token, token)
|
|
|> put_session(:user_remember_me, true)
|
|
|> put_req_cookie(@remember_me_cookie, signed_token)
|
|
|> UserAuth.fetch_current_scope_for_user([])
|
|
|
|
assert conn.assigns.current_scope.user.id == user.id
|
|
assert conn.assigns.current_scope.user.authenticated_at == user.authenticated_at
|
|
assert new_token = get_session(conn, :user_token)
|
|
assert new_token != token
|
|
assert %{value: new_signed_token, max_age: max_age} = conn.resp_cookies[@remember_me_cookie]
|
|
assert new_signed_token != signed_token
|
|
assert max_age == @remember_me_cookie_max_age
|
|
end
|
|
end
|
|
|
|
describe "require_sudo_mode/2" do
|
|
test "redirects to sudo verify when sudo mode expired and TOTP enabled", %{conn: conn} do
|
|
user = user_fixture()
|
|
|
|
# Create TOTP device
|
|
{:ok, _device, _secret} =
|
|
Accounts.create_totp_device(user.id, "Test Device", NimbleTOTP.secret())
|
|
|
|
conn =
|
|
conn
|
|
|> fetch_flash()
|
|
|> assign(:current_scope, Scope.for_user(user))
|
|
|> UserAuth.require_sudo_mode([])
|
|
|
|
assert redirected_to(conn) == ~p"/users/sudo-verify"
|
|
assert Phoenix.Flash.get(conn.assigns.flash, :error) =~ "verify your identity"
|
|
end
|
|
|
|
test "redirects to TOTP enrollment when sudo mode expired and no TOTP", %{conn: conn} do
|
|
user = user_fixture(enable_totp: false)
|
|
|
|
conn =
|
|
conn
|
|
|> fetch_flash()
|
|
|> assign(:current_scope, Scope.for_user(user))
|
|
|> UserAuth.require_sudo_mode([])
|
|
|
|
assert redirected_to(conn) == ~p"/account/totp-enrollment"
|
|
assert Phoenix.Flash.get(conn.assigns.flash, :error) =~ "Two-factor authentication"
|
|
end
|
|
|
|
test "allows access when sudo mode is active", %{conn: conn} do
|
|
user = user_fixture()
|
|
{:ok, user} = Accounts.grant_sudo_mode(user)
|
|
|
|
conn =
|
|
conn
|
|
|> fetch_flash()
|
|
|> assign(:current_scope, Scope.for_user(user))
|
|
|> UserAuth.require_sudo_mode([])
|
|
|
|
refute conn.halted
|
|
end
|
|
|
|
test "stores return path on GET requests when redirecting", %{conn: conn} do
|
|
user = user_fixture()
|
|
|
|
# Create TOTP device
|
|
{:ok, _device, _secret} =
|
|
Accounts.create_totp_device(user.id, "Test Device", NimbleTOTP.secret())
|
|
|
|
conn =
|
|
%{conn | path_info: ["admin", "users"], query_string: "", method: "GET"}
|
|
|> fetch_flash()
|
|
|> assign(:current_scope, Scope.for_user(user))
|
|
|> UserAuth.require_sudo_mode([])
|
|
|
|
assert redirected_to(conn) == ~p"/users/sudo-verify"
|
|
assert get_session(conn, :user_return_to) == "/admin/users"
|
|
end
|
|
end
|
|
|
|
describe "redirect_if_user_is_authenticated/2" do
|
|
setup %{conn: conn} do
|
|
%{conn: UserAuth.fetch_current_scope_for_user(conn, [])}
|
|
end
|
|
|
|
test "redirects if user is authenticated", %{conn: conn, user: user} do
|
|
conn =
|
|
conn
|
|
|> assign(:current_scope, Scope.for_user(user))
|
|
|> UserAuth.redirect_if_user_is_authenticated([])
|
|
|
|
assert conn.halted
|
|
assert redirected_to(conn) == ~p"/orgs"
|
|
end
|
|
|
|
test "does not redirect if user is not authenticated", %{conn: conn} do
|
|
conn = UserAuth.redirect_if_user_is_authenticated(conn, [])
|
|
refute conn.halted
|
|
refute conn.status
|
|
end
|
|
end
|
|
|
|
describe "require_authenticated_user/2" do
|
|
setup %{conn: conn} do
|
|
%{conn: UserAuth.fetch_current_scope_for_user(conn, [])}
|
|
end
|
|
|
|
test "redirects if user is not authenticated", %{conn: conn} do
|
|
conn = conn |> fetch_flash() |> UserAuth.require_authenticated_user([])
|
|
assert conn.halted
|
|
|
|
assert redirected_to(conn) == ~p"/users/log-in"
|
|
|
|
assert Phoenix.Flash.get(conn.assigns.flash, :error) ==
|
|
"You must log in to access this page."
|
|
end
|
|
|
|
test "stores the path to redirect to on GET", %{conn: conn} do
|
|
halted_conn =
|
|
%{conn | path_info: ["foo"], query_string: ""}
|
|
|> fetch_flash()
|
|
|> UserAuth.require_authenticated_user([])
|
|
|
|
assert halted_conn.halted
|
|
assert get_session(halted_conn, :user_return_to) == "/foo"
|
|
|
|
halted_conn =
|
|
%{conn | path_info: ["foo"], query_string: "bar=baz"}
|
|
|> fetch_flash()
|
|
|> UserAuth.require_authenticated_user([])
|
|
|
|
assert halted_conn.halted
|
|
assert get_session(halted_conn, :user_return_to) == "/foo?bar=baz"
|
|
|
|
halted_conn =
|
|
%{conn | path_info: ["foo"], query_string: "bar", method: "POST"}
|
|
|> fetch_flash()
|
|
|> UserAuth.require_authenticated_user([])
|
|
|
|
assert halted_conn.halted
|
|
refute get_session(halted_conn, :user_return_to)
|
|
end
|
|
|
|
test "does not redirect if user is authenticated", %{conn: conn, user: user} do
|
|
conn =
|
|
conn
|
|
|> assign(:current_scope, Scope.for_user(user))
|
|
|> UserAuth.require_authenticated_user([])
|
|
|
|
refute conn.halted
|
|
refute conn.status
|
|
end
|
|
end
|
|
|
|
describe "require_superuser/2" do
|
|
test "allows superusers", %{conn: conn} do
|
|
superuser =
|
|
user_fixture()
|
|
|> Ecto.Changeset.change(%{is_superuser: true})
|
|
|> Towerops.Repo.update!()
|
|
|
|
conn =
|
|
conn
|
|
|> fetch_flash()
|
|
|> assign(:current_scope, Scope.for_user(superuser))
|
|
|> UserAuth.require_superuser([])
|
|
|
|
refute conn.halted
|
|
end
|
|
|
|
test "redirects non-superusers", %{conn: conn, user: user} do
|
|
conn =
|
|
conn
|
|
|> fetch_flash()
|
|
|> assign(:current_scope, Scope.for_user(user))
|
|
|> UserAuth.require_superuser([])
|
|
|
|
assert conn.halted
|
|
assert redirected_to(conn) == ~p"/orgs"
|
|
|
|
assert Phoenix.Flash.get(conn.assigns.flash, :error) ==
|
|
"You must be a superuser to access this page."
|
|
end
|
|
|
|
test "redirects if not authenticated", %{conn: conn} do
|
|
conn =
|
|
conn
|
|
|> fetch_flash()
|
|
|> assign(:current_scope, Scope.for_user(nil))
|
|
|> UserAuth.require_superuser([])
|
|
|
|
assert conn.halted
|
|
assert redirected_to(conn) == ~p"/orgs"
|
|
end
|
|
end
|
|
|
|
describe "load_current_organization/2" do
|
|
test "loads organization for member", %{conn: conn, user: user} do
|
|
{:ok, organization} = Towerops.Organizations.create_organization(%{name: "Test Org"}, user.id)
|
|
|
|
conn =
|
|
conn
|
|
|> fetch_flash()
|
|
|> assign(:current_scope, Scope.for_user(user))
|
|
|> Map.put(:path_params, %{"org_slug" => organization.slug})
|
|
|> UserAuth.load_current_organization([])
|
|
|
|
refute conn.halted
|
|
assert conn.assigns.current_scope.organization.id == organization.id
|
|
assert conn.assigns.current_membership
|
|
end
|
|
|
|
test "redirects for non-member", %{conn: conn, user: user} do
|
|
other_user = user_fixture()
|
|
{:ok, organization} = Towerops.Organizations.create_organization(%{name: "Other Org"}, other_user.id)
|
|
|
|
conn =
|
|
conn
|
|
|> fetch_flash()
|
|
|> assign(:current_scope, Scope.for_user(user))
|
|
|> Map.put(:path_params, %{"org_slug" => organization.slug})
|
|
|> UserAuth.load_current_organization([])
|
|
|
|
assert conn.halted
|
|
assert redirected_to(conn) == ~p"/orgs"
|
|
|
|
assert Phoenix.Flash.get(conn.assigns.flash, :error) ==
|
|
"You don't have access to this organization."
|
|
end
|
|
|
|
test "redirects for non-existent organization", %{conn: conn, user: user} do
|
|
conn =
|
|
conn
|
|
|> fetch_flash()
|
|
|> assign(:current_scope, Scope.for_user(user))
|
|
|> Map.put(:path_params, %{"org_slug" => "non-existent"})
|
|
|> UserAuth.load_current_organization([])
|
|
|
|
assert conn.halted
|
|
assert redirected_to(conn) == ~p"/orgs"
|
|
assert Phoenix.Flash.get(conn.assigns.flash, :error) == "Organization not found."
|
|
end
|
|
|
|
test "redirects if no org_slug", %{conn: conn, user: user} do
|
|
conn =
|
|
conn
|
|
|> fetch_flash()
|
|
|> assign(:current_scope, Scope.for_user(user))
|
|
|> Map.put(:path_params, %{})
|
|
|> UserAuth.load_current_organization([])
|
|
|
|
assert conn.halted
|
|
assert redirected_to(conn) == ~p"/orgs"
|
|
end
|
|
|
|
test "redirects if no user", %{conn: conn} do
|
|
conn =
|
|
conn
|
|
|> fetch_flash()
|
|
|> assign(:current_scope, Scope.for_user(nil))
|
|
|> Map.put(:path_params, %{"org_slug" => "test"})
|
|
|> UserAuth.load_current_organization([])
|
|
|
|
assert conn.halted
|
|
assert redirected_to(conn) == ~p"/orgs"
|
|
end
|
|
end
|
|
|
|
describe "fetch_current_scope_for_user/2 with impersonation" do
|
|
test "handles valid impersonation", %{conn: conn} do
|
|
superuser = user_fixture(%{is_superuser: true})
|
|
target_user = user_fixture()
|
|
|
|
conn =
|
|
conn
|
|
|> put_session(:impersonating, true)
|
|
|> put_session(:superuser_id, superuser.id)
|
|
|> put_session(:target_user_id, target_user.id)
|
|
|> UserAuth.fetch_current_scope_for_user([])
|
|
|
|
assert conn.assigns.current_scope.user.id == target_user.id
|
|
assert conn.assigns.current_scope.superuser.id == superuser.id
|
|
end
|
|
|
|
test "clears invalid impersonation - missing superuser", %{conn: conn} do
|
|
target_user = user_fixture()
|
|
|
|
conn =
|
|
conn
|
|
|> put_session(:impersonating, true)
|
|
|> put_session(:superuser_id, Ecto.UUID.generate())
|
|
|> put_session(:target_user_id, target_user.id)
|
|
|> UserAuth.fetch_current_scope_for_user([])
|
|
|
|
refute get_session(conn, :impersonating)
|
|
refute get_session(conn, :superuser_id)
|
|
refute get_session(conn, :target_user_id)
|
|
end
|
|
|
|
test "clears invalid impersonation - missing target", %{conn: conn} do
|
|
superuser = user_fixture(%{is_superuser: true})
|
|
|
|
conn =
|
|
conn
|
|
|> put_session(:impersonating, true)
|
|
|> put_session(:superuser_id, superuser.id)
|
|
|> put_session(:target_user_id, Ecto.UUID.generate())
|
|
|> UserAuth.fetch_current_scope_for_user([])
|
|
|
|
refute get_session(conn, :impersonating)
|
|
refute get_session(conn, :superuser_id)
|
|
refute get_session(conn, :target_user_id)
|
|
end
|
|
|
|
test "clears invalid impersonation - missing IDs", %{conn: conn} do
|
|
conn =
|
|
conn
|
|
|> put_session(:impersonating, true)
|
|
|> UserAuth.fetch_current_scope_for_user([])
|
|
|
|
refute get_session(conn, :impersonating)
|
|
end
|
|
end
|
|
|
|
describe "start_impersonation/2" do
|
|
test "allows superuser to impersonate regular user", %{conn: conn} do
|
|
superuser =
|
|
user_fixture()
|
|
|> Ecto.Changeset.change(%{is_superuser: true})
|
|
|> Towerops.Repo.update!()
|
|
|
|
target_user = user_fixture()
|
|
|
|
conn =
|
|
conn
|
|
|> Map.put(:remote_ip, {127, 0, 0, 1})
|
|
|> fetch_flash()
|
|
|> assign(:current_scope, Scope.for_user(superuser))
|
|
|> UserAuth.start_impersonation(target_user.id)
|
|
|
|
assert get_session(conn, :impersonating)
|
|
assert get_session(conn, :superuser_id) == superuser.id
|
|
assert get_session(conn, :target_user_id) == target_user.id
|
|
assert conn.assigns.current_scope.user.id == target_user.id
|
|
assert conn.assigns.current_scope.superuser.id == superuser.id
|
|
assert redirected_to(conn) == ~p"/orgs"
|
|
end
|
|
|
|
test "prevents superuser from impersonating self", %{conn: conn} do
|
|
superuser =
|
|
user_fixture()
|
|
|> Ecto.Changeset.change(%{is_superuser: true})
|
|
|> Towerops.Repo.update!()
|
|
|
|
conn =
|
|
conn
|
|
|> Map.put(:remote_ip, {127, 0, 0, 1})
|
|
|> fetch_flash()
|
|
|> assign(:current_scope, Scope.for_user(superuser))
|
|
|> UserAuth.start_impersonation(superuser.id)
|
|
|
|
# Check that the error was set
|
|
assert Phoenix.Flash.get(conn.assigns.flash, :error) ==
|
|
"You cannot impersonate yourself."
|
|
|
|
assert conn.halted
|
|
end
|
|
|
|
test "allows superuser to impersonate other superusers", %{conn: conn} do
|
|
superuser =
|
|
user_fixture()
|
|
|> Ecto.Changeset.change(%{is_superuser: true})
|
|
|> Towerops.Repo.update!()
|
|
|
|
other_superuser =
|
|
user_fixture()
|
|
|> Ecto.Changeset.change(%{is_superuser: true})
|
|
|> Towerops.Repo.update!()
|
|
|
|
conn =
|
|
conn
|
|
|> Map.put(:remote_ip, {127, 0, 0, 1})
|
|
|> fetch_flash()
|
|
|> assign(:current_scope, Scope.for_user(superuser))
|
|
|> UserAuth.start_impersonation(other_superuser.id)
|
|
|
|
# Verify impersonation was successful
|
|
assert get_session(conn, :impersonating)
|
|
assert get_session(conn, :superuser_id) == superuser.id
|
|
assert get_session(conn, :target_user_id) == other_superuser.id
|
|
assert conn.assigns.current_scope.user.id == other_superuser.id
|
|
assert conn.assigns.current_scope.superuser.id == superuser.id
|
|
assert redirected_to(conn) == ~p"/orgs"
|
|
end
|
|
end
|
|
|
|
describe "stop_impersonation/1" do
|
|
test "stops impersonation and restores superuser session", %{conn: conn} do
|
|
superuser =
|
|
user_fixture()
|
|
|> Ecto.Changeset.change(%{is_superuser: true})
|
|
|> Towerops.Repo.update!()
|
|
|
|
target_user = user_fixture()
|
|
|
|
{:ok, _organization} =
|
|
Towerops.Organizations.create_organization(%{name: "Super Org"}, superuser.id)
|
|
|
|
conn =
|
|
conn
|
|
|> Map.put(:remote_ip, {127, 0, 0, 1})
|
|
|> fetch_flash()
|
|
|> put_session(:impersonating, true)
|
|
|> put_session(:superuser_id, superuser.id)
|
|
|> put_session(:target_user_id, target_user.id)
|
|
|> assign(:current_scope, Scope.for_impersonation(superuser, target_user))
|
|
|> UserAuth.stop_impersonation()
|
|
|
|
refute get_session(conn, :impersonating)
|
|
refute get_session(conn, :superuser_id)
|
|
refute get_session(conn, :target_user_id)
|
|
assert conn.assigns.current_scope.user.id == superuser.id
|
|
assert redirected_to(conn) == ~p"/dashboard"
|
|
end
|
|
|
|
test "restores superuser's organization context after stopping impersonation", %{conn: conn} do
|
|
# Create superuser with their own organization
|
|
superuser =
|
|
user_fixture()
|
|
|> Ecto.Changeset.change(%{is_superuser: true})
|
|
|> Towerops.Repo.update!()
|
|
|
|
{:ok, superuser_org} =
|
|
Towerops.Organizations.create_organization(%{name: "Superuser Org"}, superuser.id)
|
|
|
|
# Create target user with their own organization
|
|
target_user = user_fixture()
|
|
|
|
{:ok, target_org} =
|
|
Towerops.Organizations.create_organization(%{name: "Target Org"}, target_user.id)
|
|
|
|
# Simulate impersonation with target user's organization in session
|
|
conn =
|
|
conn
|
|
|> Map.put(:remote_ip, {127, 0, 0, 1})
|
|
|> fetch_flash()
|
|
|> put_session(:impersonating, true)
|
|
|> put_session(:superuser_id, superuser.id)
|
|
|> put_session(:target_user_id, target_user.id)
|
|
|> put_session(:current_organization_id, target_org.id)
|
|
|> assign(:current_scope, Scope.for_impersonation(superuser, target_user))
|
|
|> UserAuth.stop_impersonation()
|
|
|
|
# Verify organization context is restored to superuser's org
|
|
assert get_session(conn, :current_organization_id) == superuser_org.id
|
|
assert conn.assigns.current_scope.user.id == superuser.id
|
|
assert redirected_to(conn) == ~p"/dashboard"
|
|
end
|
|
|
|
test "redirects to /orgs when superuser has no organizations", %{conn: conn} do
|
|
superuser =
|
|
user_fixture()
|
|
|> Ecto.Changeset.change(%{is_superuser: true})
|
|
|> Towerops.Repo.update!()
|
|
|
|
target_user = user_fixture()
|
|
|
|
conn =
|
|
conn
|
|
|> Map.put(:remote_ip, {127, 0, 0, 1})
|
|
|> fetch_flash()
|
|
|> put_session(:impersonating, true)
|
|
|> put_session(:superuser_id, superuser.id)
|
|
|> put_session(:target_user_id, target_user.id)
|
|
|> assign(:current_scope, Scope.for_impersonation(superuser, target_user))
|
|
|> UserAuth.stop_impersonation()
|
|
|
|
assert redirected_to(conn) == ~p"/orgs"
|
|
end
|
|
end
|
|
|
|
describe "on_mount/4 - :redirect_if_user_is_authenticated" do
|
|
test "redirects authenticated users to signed in path", %{user: user} do
|
|
user_token = Accounts.generate_user_session_token(user)
|
|
|
|
{:halt, socket} =
|
|
UserAuth.on_mount(
|
|
:redirect_if_user_is_authenticated,
|
|
%{},
|
|
%{"user_token" => user_token},
|
|
%Socket{}
|
|
)
|
|
|
|
assert socket.redirected
|
|
end
|
|
|
|
test "redirects to /devices when user has organizations", %{user: user} do
|
|
{:ok, _organization} = Towerops.Organizations.create_organization(%{name: "Test Org"}, user.id)
|
|
user_token = Accounts.generate_user_session_token(user)
|
|
|
|
{:halt, socket} =
|
|
UserAuth.on_mount(
|
|
:redirect_if_user_is_authenticated,
|
|
%{},
|
|
%{"user_token" => user_token},
|
|
%Socket{}
|
|
)
|
|
|
|
assert socket.redirected
|
|
end
|
|
|
|
test "continues for unauthenticated users" do
|
|
{:cont, socket} =
|
|
UserAuth.on_mount(
|
|
:redirect_if_user_is_authenticated,
|
|
%{},
|
|
%{},
|
|
%Socket{}
|
|
)
|
|
|
|
refute socket.redirected
|
|
end
|
|
end
|
|
|
|
describe "on_mount/4 - :require_authenticated_user" do
|
|
test "allows authenticated users", %{user: user} do
|
|
user_token = Accounts.generate_user_session_token(user)
|
|
|
|
socket = %Socket{assigns: %{__changed__: %{}, flash: %{}}}
|
|
|
|
{:cont, socket} =
|
|
UserAuth.on_mount(
|
|
:require_authenticated_user,
|
|
%{},
|
|
%{"user_token" => user_token},
|
|
socket
|
|
)
|
|
|
|
assert socket.assigns.current_scope.user.id == user.id
|
|
end
|
|
|
|
test "redirects unauthenticated users to login" do
|
|
socket = %Socket{assigns: %{__changed__: %{}, flash: %{}}}
|
|
|
|
{:halt, socket} =
|
|
UserAuth.on_mount(
|
|
:require_authenticated_user,
|
|
%{},
|
|
%{},
|
|
socket
|
|
)
|
|
|
|
assert socket.redirected
|
|
end
|
|
end
|
|
|
|
describe "on_mount/4 - :load_current_organization" do
|
|
test "loads organization for authenticated member", %{user: user} do
|
|
{:ok, organization} = Towerops.Organizations.create_organization(%{name: "Test Org"}, user.id)
|
|
user_token = Accounts.generate_user_session_token(user)
|
|
|
|
socket = %Socket{assigns: %{__changed__: %{}, flash: %{}}}
|
|
|
|
socket =
|
|
Phoenix.Component.assign_new(socket, :current_scope, fn -> Scope.for_user(user) end)
|
|
|
|
{:cont, socket} =
|
|
UserAuth.on_mount(
|
|
:load_current_organization,
|
|
%{"org_slug" => organization.slug},
|
|
%{"user_token" => user_token},
|
|
socket
|
|
)
|
|
|
|
assert socket.assigns.current_scope.organization.id == organization.id
|
|
assert socket.assigns.current_membership.role == :owner
|
|
end
|
|
|
|
test "redirects non-members", %{user: user} do
|
|
other_user = user_fixture()
|
|
|
|
{:ok, organization} =
|
|
Towerops.Organizations.create_organization(%{name: "Other Org"}, other_user.id)
|
|
|
|
user_token = Accounts.generate_user_session_token(user)
|
|
|
|
socket = %Socket{assigns: %{__changed__: %{}, flash: %{}}}
|
|
|
|
socket =
|
|
Phoenix.Component.assign_new(socket, :current_scope, fn -> Scope.for_user(user) end)
|
|
|
|
{:halt, socket} =
|
|
UserAuth.on_mount(
|
|
:load_current_organization,
|
|
%{"org_slug" => organization.slug},
|
|
%{"user_token" => user_token},
|
|
socket
|
|
)
|
|
|
|
assert socket.redirected
|
|
end
|
|
|
|
test "redirects when organization not found", %{user: user} do
|
|
user_token = Accounts.generate_user_session_token(user)
|
|
|
|
socket = %Socket{assigns: %{__changed__: %{}, flash: %{}}}
|
|
|
|
socket =
|
|
Phoenix.Component.assign_new(socket, :current_scope, fn -> Scope.for_user(user) end)
|
|
|
|
{:halt, socket} =
|
|
UserAuth.on_mount(
|
|
:load_current_organization,
|
|
%{"org_slug" => "non-existent"},
|
|
%{"user_token" => user_token},
|
|
socket
|
|
)
|
|
|
|
assert socket.redirected
|
|
end
|
|
|
|
test "continues when no org_slug in params" do
|
|
socket = %Socket{assigns: %{__changed__: %{}, flash: %{}}}
|
|
|
|
{:cont, socket} =
|
|
UserAuth.on_mount(
|
|
:load_current_organization,
|
|
%{},
|
|
%{},
|
|
socket
|
|
)
|
|
|
|
refute socket.redirected
|
|
end
|
|
|
|
test "redirects when org_slug exists but no user" do
|
|
socket = %Socket{assigns: %{__changed__: %{}, flash: %{}}}
|
|
|
|
socket =
|
|
Phoenix.Component.assign_new(socket, :current_scope, fn -> Scope.for_user(nil) end)
|
|
|
|
{:halt, socket} =
|
|
UserAuth.on_mount(
|
|
:load_current_organization,
|
|
%{"org_slug" => "test-org"},
|
|
%{},
|
|
socket
|
|
)
|
|
|
|
assert socket.redirected
|
|
end
|
|
end
|
|
|
|
describe "on_mount/4 - :require_superuser" do
|
|
test "allows superusers" do
|
|
superuser =
|
|
user_fixture()
|
|
|> Ecto.Changeset.change(%{is_superuser: true})
|
|
|> Towerops.Repo.update!()
|
|
|
|
user_token = Accounts.generate_user_session_token(superuser)
|
|
|
|
socket = %Socket{assigns: %{__changed__: %{}, flash: %{}}}
|
|
|
|
{:cont, socket} =
|
|
UserAuth.on_mount(
|
|
:require_superuser,
|
|
%{},
|
|
%{"user_token" => user_token},
|
|
socket
|
|
)
|
|
|
|
assert socket.assigns.current_scope.user.id == superuser.id
|
|
end
|
|
|
|
test "redirects regular users", %{user: user} do
|
|
user_token = Accounts.generate_user_session_token(user)
|
|
|
|
socket = %Socket{assigns: %{__changed__: %{}, flash: %{}}}
|
|
|
|
{:halt, socket} =
|
|
UserAuth.on_mount(
|
|
:require_superuser,
|
|
%{},
|
|
%{"user_token" => user_token},
|
|
socket
|
|
)
|
|
|
|
assert socket.redirected
|
|
end
|
|
|
|
test "redirects unauthenticated users" do
|
|
socket = %Socket{assigns: %{__changed__: %{}, flash: %{}}}
|
|
|
|
{:halt, socket} =
|
|
UserAuth.on_mount(
|
|
:require_superuser,
|
|
%{},
|
|
%{},
|
|
socket
|
|
)
|
|
|
|
assert socket.redirected
|
|
end
|
|
end
|
|
|
|
describe "on_mount/4 - :require_sudo_mode" do
|
|
test "redirects to sudo verify when sudo mode expired and TOTP enabled" do
|
|
user = user_fixture()
|
|
|
|
# Create TOTP device
|
|
{:ok, _device, _secret} =
|
|
Accounts.create_totp_device(user.id, "Test Device", NimbleTOTP.secret())
|
|
|
|
# Create session with user token
|
|
token = Accounts.generate_user_session_token(user)
|
|
session = %{"user_token" => token}
|
|
|
|
socket = %Socket{assigns: %{__changed__: %{}, flash: %{}}}
|
|
|
|
assert {:halt, redirected_socket} =
|
|
UserAuth.on_mount(:require_sudo_mode, %{}, session, socket)
|
|
|
|
assert Phoenix.Flash.get(redirected_socket.assigns.flash, :error) =~ "verify your identity"
|
|
end
|
|
|
|
test "redirects to TOTP enrollment when sudo mode expired and no TOTP" do
|
|
user = user_fixture(enable_totp: false)
|
|
|
|
# Create session with user token
|
|
token = Accounts.generate_user_session_token(user)
|
|
session = %{"user_token" => token}
|
|
|
|
socket = %Socket{assigns: %{__changed__: %{}, flash: %{}}}
|
|
|
|
assert {:halt, redirected_socket} =
|
|
UserAuth.on_mount(:require_sudo_mode, %{}, session, socket)
|
|
|
|
assert Phoenix.Flash.get(redirected_socket.assigns.flash, :error) =~ "Two-factor authentication"
|
|
end
|
|
|
|
test "allows access when sudo mode is active" do
|
|
user = user_fixture()
|
|
{:ok, user} = Accounts.grant_sudo_mode(user)
|
|
|
|
# Create session with user token
|
|
token = Accounts.generate_user_session_token(user)
|
|
session = %{"user_token" => token}
|
|
|
|
socket = %Socket{assigns: %{__changed__: %{}, flash: %{}}}
|
|
|
|
assert {:cont, _socket} = UserAuth.on_mount(:require_sudo_mode, %{}, session, socket)
|
|
end
|
|
|
|
test "redirects unauthenticated users" do
|
|
socket = %Socket{assigns: %{__changed__: %{}, flash: %{}}}
|
|
|
|
socket =
|
|
Phoenix.Component.assign_new(socket, :current_scope, fn -> Scope.for_user(nil) end)
|
|
|
|
{:halt, result_socket} =
|
|
UserAuth.on_mount(
|
|
:require_sudo_mode,
|
|
%{},
|
|
%{},
|
|
socket
|
|
)
|
|
|
|
assert result_socket.redirected
|
|
end
|
|
end
|
|
|
|
describe "on_mount/4 - :load_default_organization" do
|
|
test "loads organization from session for authenticated user", %{user: user} do
|
|
{:ok, organization} = Towerops.Organizations.create_organization(%{name: "Test Org"}, user.id)
|
|
|
|
socket = %Socket{assigns: %{__changed__: %{}, flash: %{}}}
|
|
|
|
socket =
|
|
Phoenix.Component.assign_new(socket, :current_scope, fn -> Scope.for_user(user) end)
|
|
|
|
{:cont, result_socket} =
|
|
UserAuth.on_mount(
|
|
:load_default_organization,
|
|
%{},
|
|
%{"current_organization_id" => organization.id},
|
|
socket
|
|
)
|
|
|
|
assert result_socket.assigns.current_scope.organization.id == organization.id
|
|
assert result_socket.assigns.current_membership
|
|
end
|
|
|
|
test "loads first organization when no org_id in session", %{user: user} do
|
|
{:ok, organization} = Towerops.Organizations.create_organization(%{name: "Test Org"}, user.id)
|
|
|
|
socket = %Socket{assigns: %{__changed__: %{}, flash: %{}}}
|
|
|
|
socket =
|
|
Phoenix.Component.assign_new(socket, :current_scope, fn -> Scope.for_user(user) end)
|
|
|
|
{:cont, result_socket} =
|
|
UserAuth.on_mount(
|
|
:load_default_organization,
|
|
%{},
|
|
%{},
|
|
socket
|
|
)
|
|
|
|
assert result_socket.assigns.current_scope.organization.id == organization.id
|
|
assert result_socket.assigns.current_membership
|
|
end
|
|
|
|
test "halts when organization not found", %{user: user} do
|
|
socket = %Socket{assigns: %{__changed__: %{}, flash: %{}}}
|
|
|
|
socket =
|
|
Phoenix.Component.assign_new(socket, :current_scope, fn -> Scope.for_user(user) end)
|
|
|
|
{:halt, result_socket} =
|
|
UserAuth.on_mount(
|
|
:load_default_organization,
|
|
%{},
|
|
%{"current_organization_id" => Ecto.UUID.generate()},
|
|
socket
|
|
)
|
|
|
|
assert result_socket.redirected
|
|
end
|
|
|
|
test "halts when user has no organizations", %{user: user} do
|
|
socket = %Socket{assigns: %{__changed__: %{}, flash: %{}}}
|
|
|
|
socket =
|
|
Phoenix.Component.assign_new(socket, :current_scope, fn -> Scope.for_user(user) end)
|
|
|
|
{:halt, result_socket} =
|
|
UserAuth.on_mount(
|
|
:load_default_organization,
|
|
%{},
|
|
%{},
|
|
socket
|
|
)
|
|
|
|
assert result_socket.redirected
|
|
end
|
|
|
|
test "halts when user is not a member of organization", %{user: user} do
|
|
other_user = user_fixture()
|
|
{:ok, organization} = Towerops.Organizations.create_organization(%{name: "Other Org"}, other_user.id)
|
|
|
|
socket = %Socket{assigns: %{__changed__: %{}, flash: %{}}}
|
|
|
|
socket =
|
|
Phoenix.Component.assign_new(socket, :current_scope, fn -> Scope.for_user(user) end)
|
|
|
|
{:halt, result_socket} =
|
|
UserAuth.on_mount(
|
|
:load_default_organization,
|
|
%{},
|
|
%{"current_organization_id" => organization.id},
|
|
socket
|
|
)
|
|
|
|
assert result_socket.redirected
|
|
end
|
|
|
|
test "continues when user is not authenticated" do
|
|
socket = %Socket{assigns: %{__changed__: %{}, flash: %{}}}
|
|
|
|
socket =
|
|
Phoenix.Component.assign_new(socket, :current_scope, fn -> Scope.for_user(nil) end)
|
|
|
|
{:cont, result_socket} =
|
|
UserAuth.on_mount(
|
|
:load_default_organization,
|
|
%{},
|
|
%{},
|
|
socket
|
|
)
|
|
|
|
refute result_socket.redirected
|
|
end
|
|
end
|
|
|
|
describe "store_return_to_for_liveview/2" do
|
|
test "stores GET path when no existing return_to", %{conn: conn} do
|
|
conn =
|
|
UserAuth.store_return_to_for_liveview(
|
|
%{conn | request_path: "/devices", method: "GET", path_info: ["devices"], query_string: ""},
|
|
[]
|
|
)
|
|
|
|
assert get_session(conn, :user_return_to) == "/devices"
|
|
end
|
|
|
|
test "overwrites existing return_to with new path", %{conn: conn} do
|
|
conn =
|
|
conn
|
|
|> put_session(:user_return_to, "/original")
|
|
|> Map.put(:request_path, "/new")
|
|
|> Map.put(:path_info, ["new"])
|
|
|> Map.put(:query_string, "")
|
|
|> Map.put(:method, "GET")
|
|
|> UserAuth.store_return_to_for_liveview([])
|
|
|
|
assert get_session(conn, :user_return_to) == "/new"
|
|
end
|
|
|
|
test "does not store POST requests", %{conn: conn} do
|
|
conn = UserAuth.store_return_to_for_liveview(%{conn | request_path: "/devices", method: "POST"}, [])
|
|
|
|
refute get_session(conn, :user_return_to)
|
|
end
|
|
|
|
test "does not store login path", %{conn: conn} do
|
|
conn = UserAuth.store_return_to_for_liveview(%{conn | request_path: "/users/log-in", method: "GET"}, [])
|
|
|
|
refute get_session(conn, :user_return_to)
|
|
end
|
|
|
|
test "does not store register path", %{conn: conn} do
|
|
conn = UserAuth.store_return_to_for_liveview(%{conn | request_path: "/users/register", method: "GET"}, [])
|
|
|
|
refute get_session(conn, :user_return_to)
|
|
end
|
|
|
|
test "does not store reset password path", %{conn: conn} do
|
|
conn = UserAuth.store_return_to_for_liveview(%{conn | request_path: "/users/reset-password", method: "GET"}, [])
|
|
|
|
refute get_session(conn, :user_return_to)
|
|
end
|
|
|
|
test "does not store confirm path", %{conn: conn} do
|
|
conn = UserAuth.store_return_to_for_liveview(%{conn | request_path: "/users/confirm", method: "GET"}, [])
|
|
|
|
refute get_session(conn, :user_return_to)
|
|
end
|
|
|
|
test "does not store dev paths", %{conn: conn} do
|
|
conn = UserAuth.store_return_to_for_liveview(%{conn | request_path: "/dev/mailbox", method: "GET"}, [])
|
|
|
|
refute get_session(conn, :user_return_to)
|
|
end
|
|
|
|
test "does not store asset paths", %{conn: conn} do
|
|
conn = UserAuth.store_return_to_for_liveview(%{conn | request_path: "/assets/app.css", method: "GET"}, [])
|
|
|
|
refute get_session(conn, :user_return_to)
|
|
end
|
|
|
|
test "does not store health check paths", %{conn: conn} do
|
|
conn = UserAuth.store_return_to_for_liveview(%{conn | request_path: "/health", method: "GET"}, [])
|
|
|
|
refute get_session(conn, :user_return_to)
|
|
end
|
|
end
|
|
|
|
describe "log_out_user/1 with live_socket_id" do
|
|
test "broadcasts disconnect when live_socket_id exists", %{conn: conn, user: user} do
|
|
user_token = Accounts.generate_user_session_token(user)
|
|
live_socket_id = "users_sessions:#{user.id}"
|
|
|
|
conn =
|
|
conn
|
|
|> put_session(:user_token, user_token)
|
|
|> put_session(:live_socket_id, live_socket_id)
|
|
|> fetch_cookies()
|
|
|
|
# Subscribe to the topic to verify broadcast
|
|
ToweropsWeb.Endpoint.subscribe(live_socket_id)
|
|
|
|
conn = UserAuth.log_out_user(conn)
|
|
|
|
# Verify broadcast was sent
|
|
assert_receive %Phoenix.Socket.Broadcast{
|
|
topic: ^live_socket_id,
|
|
event: "disconnect"
|
|
}
|
|
|
|
refute get_session(conn, :user_token)
|
|
assert redirected_to(conn) == ~p"/users/log-in"
|
|
end
|
|
end
|
|
|
|
describe "redirect_if_user_is_authenticated/2 with organizations" do
|
|
test "redirects to /dashboard when user has organizations", %{conn: conn, user: user} do
|
|
{:ok, _organization} = Towerops.Organizations.create_organization(%{name: "Test Org"}, user.id)
|
|
|
|
conn =
|
|
conn
|
|
|> assign(:current_scope, Scope.for_user(user))
|
|
|> UserAuth.redirect_if_user_is_authenticated([])
|
|
|
|
assert conn.halted
|
|
assert redirected_to(conn) == ~p"/dashboard"
|
|
end
|
|
end
|
|
|
|
describe "mount_current_scope/2 timezone handling" do
|
|
test "uses user's timezone when set", %{user: user} do
|
|
user =
|
|
user
|
|
|> Ecto.Changeset.change(timezone: "America/New_York")
|
|
|> Towerops.Repo.update!()
|
|
|
|
user_token = Accounts.generate_user_session_token(user)
|
|
|
|
socket = %Socket{assigns: %{__changed__: %{}, flash: %{}}}
|
|
|
|
{:cont, result_socket} =
|
|
UserAuth.on_mount(
|
|
:require_authenticated_user,
|
|
%{},
|
|
%{"user_token" => user_token},
|
|
socket
|
|
)
|
|
|
|
assert result_socket.assigns.timezone == "America/New_York"
|
|
end
|
|
|
|
test "defaults to UTC when user has no timezone", %{user: user} do
|
|
user_token = Accounts.generate_user_session_token(user)
|
|
|
|
socket = %Socket{assigns: %{__changed__: %{}, flash: %{}}}
|
|
|
|
{:cont, result_socket} =
|
|
UserAuth.on_mount(
|
|
:require_authenticated_user,
|
|
%{},
|
|
%{"user_token" => user_token},
|
|
socket
|
|
)
|
|
|
|
# When user has no timezone and no connect_params, defaults to UTC
|
|
assert result_socket.assigns.timezone == "UTC"
|
|
end
|
|
|
|
test "defaults to UTC when not authenticated" do
|
|
socket = %Socket{assigns: %{__changed__: %{}, flash: %{}}}
|
|
|
|
{:halt, result_socket} =
|
|
UserAuth.on_mount(
|
|
:require_authenticated_user,
|
|
%{},
|
|
%{},
|
|
socket
|
|
)
|
|
|
|
# Unauthenticated users also get UTC as default
|
|
assert result_socket.assigns.timezone == "UTC"
|
|
end
|
|
end
|
|
|
|
describe "fetch_current_scope_for_user/2 edge cases" do
|
|
test "assigns nil scope when no token in session or cookies", %{conn: conn} do
|
|
conn =
|
|
conn
|
|
|> fetch_cookies()
|
|
|> UserAuth.fetch_current_scope_for_user([])
|
|
|
|
assert Map.has_key?(conn.assigns, :current_scope)
|
|
assert conn.assigns[:current_scope] == nil || conn.assigns.current_scope.user == nil
|
|
end
|
|
end
|
|
|
|
describe "on_mount/4 impersonation session building" do
|
|
test "builds impersonation scope from session" do
|
|
superuser = user_fixture(%{is_superuser: true})
|
|
target_user = user_fixture()
|
|
|
|
socket = %Socket{assigns: %{__changed__: %{}, flash: %{}}}
|
|
|
|
{:cont, result_socket} =
|
|
UserAuth.on_mount(
|
|
:require_authenticated_user,
|
|
%{},
|
|
%{
|
|
"impersonating" => true,
|
|
"superuser_id" => superuser.id,
|
|
"target_user_id" => target_user.id
|
|
},
|
|
socket
|
|
)
|
|
|
|
assert result_socket.assigns.current_scope.user.id == target_user.id
|
|
assert result_socket.assigns.current_scope.superuser.id == superuser.id
|
|
end
|
|
|
|
test "clears impersonation when superuser not found" do
|
|
target_user = user_fixture()
|
|
|
|
socket = %Socket{assigns: %{__changed__: %{}, flash: %{}}}
|
|
|
|
{:halt, result_socket} =
|
|
UserAuth.on_mount(
|
|
:require_authenticated_user,
|
|
%{},
|
|
%{
|
|
"impersonating" => true,
|
|
"superuser_id" => Ecto.UUID.generate(),
|
|
"target_user_id" => target_user.id
|
|
},
|
|
socket
|
|
)
|
|
|
|
assert result_socket.redirected
|
|
end
|
|
|
|
test "clears impersonation when target user not found" do
|
|
superuser = user_fixture(%{is_superuser: true})
|
|
|
|
socket = %Socket{assigns: %{__changed__: %{}, flash: %{}}}
|
|
|
|
{:halt, result_socket} =
|
|
UserAuth.on_mount(
|
|
:require_authenticated_user,
|
|
%{},
|
|
%{
|
|
"impersonating" => true,
|
|
"superuser_id" => superuser.id,
|
|
"target_user_id" => Ecto.UUID.generate()
|
|
},
|
|
socket
|
|
)
|
|
|
|
assert result_socket.redirected
|
|
end
|
|
|
|
test "clears impersonation when missing IDs" do
|
|
socket = %Socket{assigns: %{__changed__: %{}, flash: %{}}}
|
|
|
|
{:halt, result_socket} =
|
|
UserAuth.on_mount(
|
|
:require_authenticated_user,
|
|
%{},
|
|
%{"impersonating" => true},
|
|
socket
|
|
)
|
|
|
|
assert result_socket.redirected
|
|
end
|
|
end
|
|
|
|
describe "log_in_user/3 with return_to" do
|
|
test "uses user_return_to for redirect and clears it from session", %{conn: conn, user: user} do
|
|
conn =
|
|
conn
|
|
|> put_session(:user_return_to, "/devices")
|
|
|> UserAuth.log_in_user(user)
|
|
|
|
assert redirected_to(conn) == "/devices"
|
|
# The return_to is used for redirect but cleared from session to allow new paths to be stored
|
|
assert get_session(conn, :user_return_to) == nil
|
|
end
|
|
|
|
test "does not clear session when already logged in with same user", %{conn: conn, user: user} do
|
|
# First login
|
|
conn = UserAuth.log_in_user(conn, user)
|
|
token = get_session(conn, :user_token)
|
|
|
|
# Re-login with same user (simulating session renewal)
|
|
conn =
|
|
conn
|
|
|> recycle()
|
|
|> Map.replace!(:secret_key_base, ToweropsWeb.Endpoint.config(:secret_key_base))
|
|
|> init_test_session(%{user_token: token})
|
|
|> assign(:current_scope, Scope.for_user(user))
|
|
|> put_session(:test_data, "should_persist")
|
|
|> UserAuth.log_in_user(user)
|
|
|
|
# Session data should persist when re-authenticating with same user
|
|
assert get_session(conn, :test_data) == "should_persist"
|
|
end
|
|
|
|
test "ignores invalid return_to paths and uses default", %{conn: conn, user: user} do
|
|
# Asset paths should be ignored
|
|
conn =
|
|
conn
|
|
|> put_session(:user_return_to, "/assets/app.css")
|
|
|> UserAuth.log_in_user(user)
|
|
|
|
assert redirected_to(conn) == ~p"/orgs"
|
|
|
|
# Dev paths should be ignored
|
|
conn =
|
|
conn
|
|
|> recycle()
|
|
|> Map.replace!(:secret_key_base, ToweropsWeb.Endpoint.config(:secret_key_base))
|
|
|> init_test_session(%{})
|
|
|> put_session(:user_return_to, "/dev/mailbox")
|
|
|> UserAuth.log_in_user(user)
|
|
|
|
assert redirected_to(conn) == ~p"/orgs"
|
|
|
|
# Auth paths should be ignored
|
|
conn =
|
|
conn
|
|
|> recycle()
|
|
|> Map.replace!(:secret_key_base, ToweropsWeb.Endpoint.config(:secret_key_base))
|
|
|> init_test_session(%{})
|
|
|> put_session(:user_return_to, "/users/reset-password/token123")
|
|
|> UserAuth.log_in_user(user)
|
|
|
|
assert redirected_to(conn) == ~p"/orgs"
|
|
end
|
|
|
|
test "ignores stale mailbox asset paths (regression test)", %{conn: conn, user: user} do
|
|
# This is the exact path that was causing the bug
|
|
conn =
|
|
conn
|
|
|> put_session(:user_return_to, "/dev/mailbox/assets/app.css")
|
|
|> UserAuth.log_in_user(user)
|
|
|
|
assert redirected_to(conn) == ~p"/orgs"
|
|
assert get_session(conn, :user_return_to) == nil
|
|
end
|
|
end
|
|
|
|
describe "property-based tests" do
|
|
property "store_return_to_for_liveview always stores valid paths for GET requests", %{conn: conn} do
|
|
check all(
|
|
path <- string(:alphanumeric, min_length: 1, max_length: 50),
|
|
query <- one_of([constant(""), string(:alphanumeric, min_length: 1, max_length: 20)])
|
|
) do
|
|
path = "/" <> path
|
|
path_info = String.split(path, "/", trim: true)
|
|
|
|
conn =
|
|
UserAuth.store_return_to_for_liveview(
|
|
%{conn | request_path: path, method: "GET", path_info: path_info, query_string: query},
|
|
[]
|
|
)
|
|
|
|
stored_path = get_session(conn, :user_return_to)
|
|
|
|
if String.starts_with?(path, "/users/log-in") or String.starts_with?(path, "/users/register") or
|
|
String.starts_with?(path, "/users/reset-password") or
|
|
String.starts_with?(path, "/users/confirm") do
|
|
refute stored_path
|
|
else
|
|
assert stored_path
|
|
assert String.starts_with?(stored_path, "/")
|
|
end
|
|
end
|
|
end
|
|
|
|
property "fetch_current_scope_for_user always assigns a scope", %{conn: conn} do
|
|
check all(has_token <- boolean()) do
|
|
conn =
|
|
if has_token do
|
|
user = user_fixture(%{authenticated_at: DateTime.utc_now(:second)})
|
|
token = Accounts.generate_user_session_token(user)
|
|
put_session(conn, :user_token, token)
|
|
else
|
|
conn
|
|
end
|
|
|
|
conn = UserAuth.fetch_current_scope_for_user(conn, [])
|
|
|
|
assert Map.has_key?(conn.assigns, :current_scope)
|
|
end
|
|
end
|
|
|
|
property "redirect paths are always valid for authenticated users", %{user: user} do
|
|
check all(has_org <- boolean(), max_runs: 10) do
|
|
if has_org do
|
|
{:ok, _org} = Towerops.Organizations.create_organization(%{name: "Test Org"}, user.id, bypass_limits: true)
|
|
end
|
|
|
|
# Reload user to get fresh org data
|
|
user = Towerops.Repo.reload(user)
|
|
|
|
conn =
|
|
build_conn()
|
|
|> Map.replace!(:secret_key_base, ToweropsWeb.Endpoint.config(:secret_key_base))
|
|
|> init_test_session(%{})
|
|
|> assign(:current_scope, Scope.for_user(user))
|
|
|> UserAuth.redirect_if_user_is_authenticated([])
|
|
|
|
assert conn.halted
|
|
redirect_path = redirected_to(conn)
|
|
assert redirect_path in [~p"/orgs", ~p"/dashboard"]
|
|
end
|
|
end
|
|
end
|
|
end
|