import asyncio from pathlib import Path import httpx2 from gewu.adapters.git_repository import GitCanonicalRepository from gewu.adapters.sqlite_identity import SQLiteIdentityStore from gewu.adapters.sqlite_search import SQLiteSearchIndex from gewu.application.reading import CanonicalNodeReader from gewu.application.search import CanonicalSearchService from gewu.bootstrap.app import create_app from tests.domain.proposal_fixtures import PROJECTS def test_browser_session_is_http_only_human_only_and_csrf_protected(tmp_path: Path) -> None: async def exercise() -> None: repository = GitCanonicalRepository(tmp_path / "knowledge.git") identities = SQLiteIdentityStore(tmp_path / "identity.sqlite3") viewer_key = identities.provision( credential_id="browser-viewer-key", principal_id="browser-viewer", roles=("viewer",), project_ids=("dome-performance",), ) agent_key = identities.provision( credential_id="browser-agent-key", principal_id="browser-agent", roles=("agent",), project_ids=("dome-performance",), ) search = CanonicalSearchService( index=SQLiteSearchIndex(tmp_path / "search.sqlite3", repository), project_ids_provider=lambda: PROJECTS.keys(), canonical_head=repository.head, ) app = create_app( identity_provider=identities, node_reader=CanonicalNodeReader(repository), search_service=search, ) async with httpx2.AsyncClient( transport=httpx2.ASGITransport(app=app), base_url="http://gewu.test", ) as client: agent_session = await client.post( "/v1/session", headers={ "Authorization": f"Bearer {agent_key}", "Origin": "http://gewu.test", }, ) assert agent_session.status_code == 403 cross_origin = await client.post( "/v1/session", headers={ "Authorization": f"Bearer {viewer_key}", "Origin": "https://attacker.invalid", }, ) assert cross_origin.status_code == 403 created = await client.post( "/v1/session", headers={ "Authorization": f"Bearer {viewer_key}", "Origin": "http://gewu.test", }, ) assert created.status_code == 200, created.text session = created.json() cookie = created.cookies.get("gewu_session") assert cookie is not None assert "httponly" in created.headers["set-cookie"].lower() assert "samesite=strict" in created.headers["set-cookie"].lower() assert "max-age" not in created.headers["set-cookie"].lower() projects = await client.get("/v1/projects") assert projects.status_code == 200 assert projects.json() == ["dome-performance"] denied_csrf = await client.post( "/v1/search", json={ "query": "visitors", "scope": {"projects": ["dome-performance"], "include_shared": False}, "budget_tokens": 5000, }, ) assert denied_csrf.status_code == 403 searched = await client.post( "/v1/search", headers={"X-GEWU-CSRF": session["csrf_token"]}, json={ "query": "visitors", "scope": {"projects": ["dome-performance"], "include_shared": False}, "budget_tokens": 5000, }, ) assert searched.status_code == 200, searched.text signed_out = await client.delete( "/v1/session", headers={"X-GEWU-CSRF": session["csrf_token"]} ) assert signed_out.status_code == 200 assert (await client.get("/v1/projects")).status_code == 401 assert identities.authenticate_browser_session(cookie) is None asyncio.run(exercise()) def test_session_restoration_preserves_other_tabs_and_original_expiry(tmp_path: Path) -> None: async def exercise() -> None: identities = SQLiteIdentityStore(tmp_path / "identity.sqlite3") token = identities.provision( credential_id="admin", principal_id="operator", roles=("admin",) ) app = create_app( identity_provider=identities, node_reader=CanonicalNodeReader(GitCanonicalRepository(tmp_path / "knowledge.git")), ) async with httpx2.AsyncClient( transport=httpx2.ASGITransport(app=app), base_url="https://gewu.test" ) as http: headers = {"X-GEWU-Session": "restore", "Origin": "https://gewu.test"} assert (await http.get("/v1/session", headers=headers)).status_code == 401 login = await http.post("/v1/session", headers={"Authorization": f"Bearer {token}"}) original_csrf = login.json()["csrf_token"] cookie = login.cookies.get("gewu_session") assert cookie assert (await http.get("/v1/session")).status_code == 403 assert ( await http.get( "/v1/session", headers={**headers, "Origin": "https://attacker.invalid"} ) ).status_code == 403 assert ( await http.get("/v1/session", headers={**headers, "Sec-Fetch-Site": "cross-site"}) ).status_code == 403 restored = await http.get("/v1/session", headers=headers) assert restored.status_code == 200 assert restored.json()["expires_at"] == login.json()["expires_at"] assert restored.json()["password_account"] is False assert restored.headers["cache-control"] == "private, no-store" assert "set-cookie" not in restored.headers assert cookie not in restored.text and token not in restored.text restored_csrf = restored.json()["csrf_token"] assert identities.browser_csrf_valid(cookie, original_csrf) assert identities.browser_csrf_valid(cookie, restored_csrf) second = await http.get("/v1/session", headers=headers) assert second.json()["csrf_token"] == restored_csrf assert identities.browser_csrf_valid(cookie, original_csrf) assert not identities.browser_csrf_valid(cookie, "gwc_invalid") signed_out = await http.delete("/v1/session", headers={"X-GEWU-CSRF": restored_csrf}) assert signed_out.status_code == 200 http.cookies.set("gewu_session", cookie) assert (await http.get("/v1/session", headers=headers)).status_code == 401 next_login = identities.create_browser_session(token) http.cookies.clear() http.cookies.set("gewu_session", next_login.session_token) identities.revoke("admin") assert (await http.get("/v1/session", headers=headers)).status_code == 401 asyncio.run(exercise()) def test_password_session_restore_and_pre_upgrade_session_migration(tmp_path: Path) -> None: identities = SQLiteIdentityStore(tmp_path / "identity.sqlite3") identities.create_account( username="admin", password="fixture-password", role="admin", project_ids=(), actor="operator", ) session = identities.login_account("admin", "fixture-password") assert session # Reconstruct the pre-upgrade schema with an issued session, then run the migration. with identities._connect() as db: db.execute("ALTER TABLE browser_sessions DROP COLUMN restored_csrf_digest") reopened = SQLiteIdentityStore(identities.database) restored = reopened.restore_browser_session(session.session_token) assert restored and restored.expires_at == session.expires_at assert reopened.browser_csrf_valid(session.session_token, session.csrf_token) assert reopened.browser_csrf_valid(session.session_token, restored.csrf_token) with reopened._connect() as db: db.execute("UPDATE browser_sessions SET expires_at=0") assert reopened.restore_browser_session(session.session_token) is None