407db57279
Both password and WebAuthn login wrote the authenticated identity onto the existing pre-auth session, so a fixed/planted session could be elevated to an authenticated one. Add _establish_authenticated_session() which clears the session (preserving only a pending OIDC authorization request) before setting the identity, used by both login paths. Tests that reused a pre-login CSRF token now re-fetch it after login, matching real client behavior. Refs: porchlight-vxr Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
87 lines
2.8 KiB
Python
87 lines
2.8 KiB
Python
from datetime import UTC, datetime
|
|
|
|
import pytest
|
|
from httpx import AsyncClient
|
|
|
|
from porchlight.authn.password import PasswordHasher, PasswordService
|
|
from porchlight.models import PasswordCredential, User
|
|
from tests.conftest import get_csrf_token
|
|
|
|
|
|
async def _setup_admin_and_target(client: AsyncClient) -> tuple[str, str]:
|
|
"""Create admin + target user, login as admin, return (token, target_userid)."""
|
|
app = client._transport.app
|
|
user_repo = app.state.user_repo
|
|
cred_repo = app.state.credential_repo
|
|
|
|
admin = User(
|
|
userid="admin-g01",
|
|
username="admin_g",
|
|
groups=["admin", "users"],
|
|
created_at=datetime.now(UTC),
|
|
updated_at=datetime.now(UTC),
|
|
)
|
|
await user_repo.create(admin)
|
|
|
|
target = User(
|
|
userid="target-g01",
|
|
username="target_g",
|
|
groups=["users"],
|
|
created_at=datetime.now(UTC),
|
|
updated_at=datetime.now(UTC),
|
|
)
|
|
await user_repo.create(target)
|
|
|
|
svc = PasswordService(hasher=PasswordHasher(time_cost=1, memory_cost=8192))
|
|
await cred_repo.create_password(PasswordCredential(user_id=admin.userid, password_hash=svc.hash("AdminPass123!")))
|
|
|
|
token = await get_csrf_token(client)
|
|
await client.post(
|
|
"/login/password",
|
|
data={"username": "admin_g", "password": "AdminPass123!"},
|
|
headers={"HX-Request": "true", "X-CSRF-Token": token},
|
|
)
|
|
# Login resets the session (fixation defense); fetch a fresh CSRF token.
|
|
token = await get_csrf_token(client)
|
|
return token, target.userid
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_valid_groups(client: AsyncClient) -> None:
|
|
token, userid = await _setup_admin_and_target(client)
|
|
response = await client.post(
|
|
f"/admin/users/{userid}/groups",
|
|
data={"groups": "users, staff"},
|
|
headers={"X-CSRF-Token": token},
|
|
)
|
|
assert "Groups updated" in response.text
|
|
|
|
app = client._transport.app
|
|
user = await app.state.user_repo.get_by_userid(userid)
|
|
assert sorted(user.groups) == ["staff", "users"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_invalid_group_name_rejected(client: AsyncClient) -> None:
|
|
token, userid = await _setup_admin_and_target(client)
|
|
response = await client.post(
|
|
f"/admin/users/{userid}/groups",
|
|
data={"groups": "users, Bad Group!"},
|
|
headers={"X-CSRF-Token": token},
|
|
)
|
|
assert "alert" in response.text
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_empty_groups_clears(client: AsyncClient) -> None:
|
|
token, userid = await _setup_admin_and_target(client)
|
|
response = await client.post(
|
|
f"/admin/users/{userid}/groups",
|
|
data={"groups": ""},
|
|
headers={"X-CSRF-Token": token},
|
|
)
|
|
assert "Groups updated" in response.text
|
|
|
|
app = client._transport.app
|
|
user = await app.state.user_repo.get_by_userid(userid)
|
|
assert user.groups == []
|