190 lines
8.3 KiB
Python
190 lines
8.3 KiB
Python
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
|