| Policy Name |
|---|
{KEY_ID}")), "{body}");
+ assert!(
+ body.contains(r#"action="/console/accounts/444455556666/users/bob/access-keys/new""#),
+ "{body}"
+ );
+ assert!(
+ body.contains(&format!(
+ r#"action="/console/accounts/444455556666/users/bob/access-keys/{KEY_ID}/delete""#
+ )),
+ "{body}"
+ );
+}
+
+#[tokio::test]
+async fn iam_user_page_leaves_out_policies_groups_and_tags() {
+ let response = render_user_page(&session(iam_user()), ACCOUNT, USER, Some(&detail()));
+
+ let body = body_text(response).await;
+ for absent in [
+ POLICY,
+ GROUP,
+ TAG_KEY,
+ TAG_VALUE,
+ "Accounts
" not in page.text - assert f'' in page.text + assert f'' in page.text for name in world.a.names() + [world.a.id]: assert name not in page.text, name diff --git a/tests/test_console_iam_user_pages.py b/tests/test_console_iam_user_pages.py new file mode 100644 index 000000000..7230d9794 --- /dev/null +++ b/tests/test_console_iam_user_pages.py @@ -0,0 +1,600 @@ +# Copyright 2026 ExtendDB contributors +# SPDX-License-Identifier: Apache-2.0 + +"""Console pages of an IAM user's own account serve the user its own user +page and its own access keys only. + +The management API gives an IAM user its own access keys (list, create and +delete) and its own password, and answers every other read and write with +HTTP 403. The console gives an IAM-user session the same: its own user page, +which lists its access keys, and the forms that create and delete them. +Every other page and form under its account redirects it to /console, and no +page it can open names another user, group or role, or links to a page that +redirects it. + +The account holds a user `owner` with an inline policy, an access key and a +tag; a second user `peer`; a group with `owner` as its member and an inline +policy; and a role with an inline policy. The IAM user `viewer` is created +with a console password, so it holds only the self-service policy that every +such user gets. It is the member of a second group and has a tag, which its +own page leaves out. Each name carries a random suffix, so that no page +shows it by chance. + +Prerequisites match tests/test_console_account_scope.py: + - extenddb running with `auth.provider = "builtin"` on EXTENDDB_TEST_ENDPOINT + - Admin credentials in EXTENDDB_ADMIN_USER / EXTENDDB_ADMIN_PASSWORD +""" + +from __future__ import annotations + +import json +import os +import re +import uuid +from dataclasses import dataclass +from typing import Callable + +import pytest +import requests + +from management_helpers import ManagementClient + +_CSRF_RE = re.compile(r'{world.viewer_key}" in resp.text
+ assert f'action="/console/accounts/{a}/users/{user}/access-keys/new"' in resp.text
+ assert (f'action="/console/accounts/{a}/users/{user}/access-keys/'
+ f'{world.viewer_key}/delete"') in resp.text
+ # The management API refuses the viewer its own policies and tags, so
+ # the page leaves out the policies, groups and tags sections.
+ for absent in ("{key_id}" in page.text
+
+ deleted = world.viewer.post(f"/accounts/{a}/users/{user}/access-keys/{key_id}/delete", {})
+ assert deleted.status_code == 303, deleted.status_code
+ assert deleted.headers.get("location") == f"/console/accounts/{a}/users/{user}"
+ assert key_id not in world.mgmt.list_access_keys(a, user).text
+ assert world.viewer_key in world.mgmt.list_access_keys(a, user).text
+
+
+def _viewer_key_ids(w: World) -> set[str]:
+ """The viewer's access key IDs, as the management API lists them."""
+ resp = w.mgmt.list_access_keys(w.account_id, w.viewer_name)
+ assert resp.status_code == 200, resp.text
+ return {key["access_key_id"] for key in resp.json()}
+
+
+# A `_csrf` value other than the session's token: an empty one, and one of
+# the same length that differs from the token in its last character.
+WRONG_CSRF: dict[str, Callable[[str], str]] = {
+ "empty": lambda token: "",
+ "altered": lambda token: token[:-1] + ("1" if token.endswith("0") else "0"),
+}
+
+
+@pytest.mark.parametrize("csrf", sorted(WRONG_CSRF))
+@pytest.mark.parametrize("form", ["create-key", "delete-key"])
+def test_key_form_with_a_wrong_csrf_token_gets_403_and_changes_no_key(
+ world: World, form: str, csrf: str
+) -> None:
+ """The two forms an IAM-user session can submit check the session's
+ CSRF token before they change a key."""
+ a, user = world.account_id, world.viewer_name
+ # A key for the delete form to name, so that the viewer's own key stays
+ # in place if the form deletes it.
+ extra = world.mgmt.create_access_key(a, user)
+ assert extra.status_code == 201, extra.text
+ path = {
+ "create-key": f"/accounts/{a}/users/{user}/access-keys/new",
+ "delete-key": (f"/accounts/{a}/users/{user}/access-keys/"
+ f"{extra.json()['access_key_id']}/delete"),
+ }[form]
+ before = _viewer_key_ids(world)
+ try:
+ resp = world.viewer.post(path, {"_csrf": WRONG_CSRF[csrf](world.viewer.csrf)})
+ after = _viewer_key_ids(world)
+ finally:
+ for key_id in _viewer_key_ids(world) - {world.viewer_key}:
+ world.mgmt.delete_access_key(a, user, key_id)
+
+ assert resp.status_code == 403, f"{form}: {resp.status_code} {resp.text[:300]}"
+ assert after == before, f"{form} changed the keys from {before} to {after}"
+
+
+def _viewer_pages(world: World) -> dict[str, str]:
+ """Every page the viewer opens: the home page, its user page, and the
+ page that shows a key it creates. The key is deleted again."""
+ a, user = world.account_id, world.viewer_name
+ pages = {}
+ for name, path in (("home", ""), ("user", f"/accounts/{a}/users/{user}")):
+ resp = world.viewer.get(path)
+ assert resp.status_code == 200, f"{name}: {resp.status_code}"
+ pages[name] = resp.text
+ created = world.viewer.post(f"/accounts/{a}/users/{user}/access-keys/new", {})
+ assert created.status_code == 200, created.status_code
+ pages["new-key"] = created.text
+ match = _KEY_RE.search(created.text)
+ assert match, created.text[:500]
+ deleted = world.viewer.post(
+ f"/accounts/{a}/users/{user}/access-keys/{match.group(1)}/delete", {})
+ assert deleted.status_code == 303, deleted.status_code
+ return pages
+
+
+def test_no_page_the_viewer_opens_names_another_principal(world: World) -> None:
+ for page, text in _viewer_pages(world).items():
+ for name in world.others():
+ assert name not in text, f"the {page} page shows {name}"
+
+
+def test_viewer_pages_link_to_no_page_that_redirects_it(world: World) -> None:
+ pages = _viewer_pages(world)
+ own_page = f"/console/accounts/{world.account_id}/users/{world.viewer_name}"
+ for page in ("home", "new-key"):
+ assert f'href="{own_page}"' in pages[page], f"the {page} page has no link to {own_page}"
+ # The Metrics, Settings and Cache links are outside the account pages,
+ # and the documentation pages are public, so the check covers the home
+ # page and the pages under /console/accounts.
+ for page, text in pages.items():
+ for link in sorted(set(_HREF_RE.findall(text))):
+ if link != "/console" and not link.startswith("/console/accounts"):
+ continue
+ resp = world.viewer.open(link)
+ assert resp.status_code == 200, (
+ f"the {page} page links to {link}, which answers {resp.status_code} "
+ f"{resp.headers.get('location')}")
+
+
+def test_viewer_logs_in_manages_keys_and_logs_out(world: World) -> None:
+ a, user = world.account_id, world.viewer_name
+ browser = Console(world.endpoint)
+ home = browser.login(*world.viewer_auth())
+ own_page = f"/console/accounts/{a}/users/{user}"
+ assert f'' in home.text
+
+ page = browser.open(own_page)
+ assert page.status_code == 200, page.status_code
+ assert f"{world.viewer_key}" in page.text
+
+ resp = browser.post("/logout", {})
+ assert resp.status_code == 303, resp.status_code
+ assert resp.headers.get("location") == "/console/login"
+ resp = browser.open(own_page)
+ assert resp.status_code == 303, resp.status_code
+ assert resp.headers.get("location") == "/console/login"
+
+
+# The names each page shows an admin.
+SHOWN: dict[str, Callable[[World], tuple[str, list[str]]]] = {
+ "accounts": lambda w: ("/accounts", [w.account_id, w.account_name]),
+ "account": lambda w: (
+ f"/accounts/{w.account_id}",
+ [w.account_name, w.owner, w.peer, w.viewer_name, w.group, w.role]),
+ "owner": lambda w: (
+ f"/accounts/{w.account_id}/users/{w.owner}",
+ [w.owner_key, w.user_policy, w.group, w.tag_key, w.tag_value]),
+ "viewer": lambda w: (
+ f"/accounts/{w.account_id}/users/{w.viewer_name}",
+ [w.viewer_key, _SELF_SERVICE_POLICY, *w.viewer_hidden()]),
+ "group": lambda w: (
+ f"/accounts/{w.account_id}/groups/{w.group}", [w.owner, w.group_policy]),
+ "role": lambda w: (f"/accounts/{w.account_id}/roles/{w.role}", [w.role_policy]),
+}
+
+
+@pytest.mark.parametrize("page", sorted(SHOWN))
+def test_admin_reads_every_page_of_the_account(world: World, page: str) -> None:
+ path, names = SHOWN[page](world)
+ resp = world.admin.get(path)
+ assert resp.status_code == 200, f"{page}: {resp.status_code}"
+ for name in names:
+ assert name in resp.text, f"{page} lacks {name}"
diff --git a/tests/test_console_integration.py b/tests/test_console_integration.py
index 994a9f6f0..d1dd21930 100755
--- a/tests/test_console_integration.py
+++ b/tests/test_console_integration.py
@@ -781,15 +781,17 @@ def test_iam_user_sees_own_account_only(self):
resp = iam_console.login(f"{self.account_id}/{user}", password)
assert resp.status_code == 303
- # Dashboard should be accessible.
+ # Dashboard should be accessible, with a link to the user's own
+ # page.
resp = iam_console.get("")
assert resp.status_code == 200
+ assert f"/console/accounts/{self.account_id}/users/{user}" in resp.text
- # Account listing: IAM user should see limited view.
+ # The account list is for admins: an IAM user is redirected to
+ # the dashboard.
resp = iam_console.get("/accounts")
- assert resp.status_code == 200
- # The IAM user's own account should be visible.
- assert self.account_id in resp.text
+ assert resp.status_code == 303
+ assert resp.headers.get("location") == "/console"
finally:
self.admin_console.post(
f"/accounts/{self.account_id}/users/{user}/delete"
diff --git a/tests/test_cross_account_isolation.py b/tests/test_cross_account_isolation.py
index cb88bc3d6..06ff1839b 100755
--- a/tests/test_cross_account_isolation.py
+++ b/tests/test_cross_account_isolation.py
@@ -356,27 +356,29 @@ def _console_login(self, account_id: str, user_name: str,
return session
def test_alice_console_does_not_show_bob_account(self):
- """Alice's console session does not list bob's account."""
+ """Alice's console session does not list bob's account: the account
+ list is for admins and redirects her to the dashboard."""
session = self._console_login(self.alice_acct, "alice", "AlicePass123!")
resp = session.get(
f"{self.endpoint}/console/accounts",
allow_redirects=False,
timeout=30,
)
- assert resp.status_code == 200
- assert self.alice_acct in resp.text
+ assert resp.status_code == 303
+ assert resp.headers.get("location") == "/console"
assert self.bob_acct not in resp.text
def test_bob_console_does_not_show_alice_account(self):
- """Bob's console session does not list alice's account."""
+ """Bob's console session does not list alice's account: the account
+ list is for admins and redirects him to the dashboard."""
session = self._console_login(self.bob_acct, "bob", "BobPass456!")
resp = session.get(
f"{self.endpoint}/console/accounts",
allow_redirects=False,
timeout=30,
)
- assert resp.status_code == 200
- assert self.bob_acct in resp.text
+ assert resp.status_code == 303
+ assert resp.headers.get("location") == "/console"
assert self.alice_acct not in resp.text