265 lines
12 KiB
Python
265 lines
12 KiB
Python
import asyncio
|
|
import sqlite3
|
|
from pathlib import Path
|
|
|
|
import httpx2
|
|
|
|
from gewu.adapters.git_repository import GitCanonicalRepository
|
|
from gewu.adapters.sqlite_identity import SQLiteIdentityStore
|
|
from gewu.application.reading import CanonicalNodeReader
|
|
from gewu.bootstrap.app import create_app
|
|
|
|
PASSWORD = "initial-password-fixture"
|
|
|
|
|
|
def test_password_accounts_owned_agent_keys_and_session_lifecycle(tmp_path: Path) -> None:
|
|
async def exercise() -> None:
|
|
store = SQLiteIdentityStore(tmp_path / "identity.sqlite3")
|
|
operator = store.provision(
|
|
credential_id="operator", principal_id="operator", roles=("admin",)
|
|
)
|
|
app = create_app(
|
|
identity_provider=store,
|
|
account_identity=store,
|
|
credential_administration=store,
|
|
node_reader=CanonicalNodeReader(GitCanonicalRepository(tmp_path / "knowledge.git")),
|
|
)
|
|
transport = httpx2.ASGITransport(app=app)
|
|
operator_headers = {"Authorization": f"Bearer {operator}"}
|
|
async with httpx2.AsyncClient(transport=transport, base_url="https://gewu.test") as admin:
|
|
body = {
|
|
"username": "alice",
|
|
"password": PASSWORD,
|
|
"role": "editor",
|
|
"project_ids": ["pilot"],
|
|
}
|
|
assert (await admin.post("/v1/admin/users", json=body)).status_code == 401
|
|
created = await admin.post("/v1/admin/users", json=body, headers=operator_headers)
|
|
assert created.status_code == 201 and PASSWORD not in created.text
|
|
assert (
|
|
await admin.post("/v1/admin/users", json=body, headers=operator_headers)
|
|
).status_code == 409
|
|
for username, role in (("bob", "editor"), ("viewer", "viewer"), ("admin", "admin")):
|
|
assert (
|
|
await admin.post(
|
|
"/v1/admin/users",
|
|
headers=operator_headers,
|
|
json={**body, "username": username, "role": role},
|
|
)
|
|
).status_code == 201
|
|
short_secret = "shortsecret"
|
|
invalid = await admin.post(
|
|
"/v1/admin/users",
|
|
headers=operator_headers,
|
|
json={**body, "username": "invalid", "password": short_secret},
|
|
)
|
|
assert invalid.status_code == 422 and short_secret not in invalid.text
|
|
listing = await admin.get("/v1/admin/users", headers=operator_headers)
|
|
assert "password_hash" not in listing.text and PASSWORD not in listing.text
|
|
|
|
async with httpx2.AsyncClient(
|
|
transport=transport, base_url="https://gewu.test"
|
|
) as alice:
|
|
login = await alice.post(
|
|
"/v1/users/session",
|
|
json={"username": "alice", "password": PASSWORD},
|
|
headers={"Origin": "https://gewu.test"},
|
|
)
|
|
assert login.status_code == 200
|
|
assert "secure" in login.headers["set-cookie"].lower()
|
|
csrf = {"X-GEWU-CSRF": login.json()["csrf_token"]}
|
|
assert (await alice.get("/v1/me")).json()["principal_id"] == "user:alice"
|
|
assert (await alice.get("/v1/admin/users")).status_code == 403
|
|
assert (
|
|
await alice.post("/v1/admin/users", headers=csrf, json=body)
|
|
).status_code == 403
|
|
agent_body = {"name": "codex-laptop", "project_ids": ["pilot"]}
|
|
assert (
|
|
await alice.post("/v1/me/agent-credentials", json=agent_body)
|
|
).status_code == 403
|
|
assert (
|
|
await alice.post(
|
|
"/v1/me/agent-credentials",
|
|
headers={**csrf, "Origin": "https://attacker"},
|
|
json=agent_body,
|
|
)
|
|
).status_code == 403
|
|
assert (
|
|
await alice.post(
|
|
"/v1/me/agent-credentials",
|
|
headers=csrf,
|
|
json={**agent_body, "project_ids": ["other"]},
|
|
)
|
|
).status_code == 403
|
|
assert (
|
|
await alice.post(
|
|
"/v1/me/agent-credentials",
|
|
headers=csrf,
|
|
json={**agent_body, "role": "admin"},
|
|
)
|
|
).status_code == 422
|
|
agent = await alice.post("/v1/me/agent-credentials", headers=csrf, json=agent_body)
|
|
assert agent.status_code == 201
|
|
key_id, token = agent.json()["credential_id"], agent.json()["token"]
|
|
agent_principal = store.authenticate(token)
|
|
assert agent_principal and agent_principal.can_submit_project("pilot")
|
|
assert not agent_principal.can_submit_project("other")
|
|
assert (await alice.get("/v1/me/agent-credentials")).json()[0][
|
|
"credential_id"
|
|
] == key_id
|
|
assert token not in (await alice.get("/v1/me/agent-credentials")).text
|
|
assert (
|
|
await alice.get("/v1/me", headers={"Authorization": f"Bearer {token}"})
|
|
).status_code == 403
|
|
|
|
async with httpx2.AsyncClient(
|
|
transport=transport, base_url="https://gewu.test"
|
|
) as bob:
|
|
bob_login = await bob.post(
|
|
"/v1/users/session", json={"username": "bob", "password": PASSWORD}
|
|
)
|
|
bob_csrf = {"X-GEWU-CSRF": bob_login.json()["csrf_token"]}
|
|
assert (await bob.get("/v1/me/agent-credentials")).json() == []
|
|
assert (
|
|
await bob.delete(f"/v1/me/agent-credentials/{key_id}", headers=bob_csrf)
|
|
).status_code == 404
|
|
assert store.authenticate(token) is not None
|
|
bob_agent = await bob.post(
|
|
"/v1/me/agent-credentials", headers=bob_csrf, json=agent_body
|
|
)
|
|
rotated = store.rotate(bob_agent.json()["credential_id"], "rotated-bob-key")
|
|
assert (await bob.get("/v1/me/agent-credentials")).json()[-1][
|
|
"owner_principal_id"
|
|
] == "user:bob"
|
|
assert (
|
|
await admin.delete(
|
|
"/v1/admin/credentials/account:bob", headers=operator_headers
|
|
)
|
|
).status_code == 200
|
|
assert store.authenticate(rotated) is None
|
|
assert (await bob.get("/v1/me")).status_code == 401
|
|
|
|
password_body = {
|
|
"current_password": PASSWORD,
|
|
"new_password": "changed-password-fixture",
|
|
}
|
|
assert (await alice.post("/v1/me/password", json=password_body)).status_code == 403
|
|
assert (
|
|
await alice.post(
|
|
"/v1/me/password",
|
|
headers=csrf,
|
|
json={**password_body, "current_password": "wrong"},
|
|
)
|
|
).status_code == 403
|
|
second_session = store.login_account("alice", PASSWORD)
|
|
assert second_session
|
|
changed = await alice.post("/v1/me/password", headers=csrf, json=password_body)
|
|
assert changed.status_code == 200
|
|
assert store.authenticate_browser_session(second_session.session_token) is None
|
|
assert (await alice.get("/v1/me")).status_code == 401
|
|
assert store.login_account("alice", PASSWORD) is None
|
|
assert store.authenticate(token) is not None
|
|
relogin = await alice.post(
|
|
"/v1/users/session",
|
|
json={"username": "alice", "password": password_body["new_password"]},
|
|
)
|
|
csrf = {"X-GEWU-CSRF": relogin.json()["csrf_token"]}
|
|
assert (
|
|
await alice.delete(f"/v1/me/agent-credentials/{key_id}", headers=csrf)
|
|
).status_code == 200
|
|
assert store.authenticate(token) is None
|
|
another = await alice.post(
|
|
"/v1/me/agent-credentials", headers=csrf, json=agent_body
|
|
)
|
|
assert (
|
|
await admin.delete("/v1/admin/users/alice", headers=operator_headers)
|
|
).status_code == 200
|
|
assert store.authenticate(another.json()["token"]) is None
|
|
assert (await alice.get("/v1/me")).status_code == 401
|
|
assert store.login_account("alice", password_body["new_password"]) is None
|
|
|
|
async with httpx2.AsyncClient(
|
|
transport=transport, base_url="https://gewu.test"
|
|
) as viewer:
|
|
login = await viewer.post(
|
|
"/v1/users/session", json={"username": "viewer", "password": PASSWORD}
|
|
)
|
|
csrf = {"X-GEWU-CSRF": login.json()["csrf_token"]}
|
|
assert (
|
|
await viewer.post(
|
|
"/v1/me/agent-credentials",
|
|
headers=csrf,
|
|
json={"name": "agent", "project_ids": ["pilot"]},
|
|
)
|
|
).status_code == 403
|
|
async with httpx2.AsyncClient(
|
|
transport=transport, base_url="https://gewu.test"
|
|
) as human_admin:
|
|
login = await human_admin.post(
|
|
"/v1/users/session", json={"username": "admin", "password": PASSWORD}
|
|
)
|
|
csrf = {"X-GEWU-CSRF": login.json()["csrf_token"]}
|
|
assert (
|
|
await human_admin.delete("/v1/admin/users/admin", headers=csrf)
|
|
).status_code == 403
|
|
with sqlite3.connect(store.database) as db:
|
|
assert (
|
|
db.execute("SELECT password_hash FROM user_accounts WHERE username='bob'")
|
|
.fetchone()[0]
|
|
.startswith("$argon2id$")
|
|
)
|
|
assert (
|
|
PASSWORD not in (await admin.get("/v1/admin/events", headers=operator_headers)).text
|
|
)
|
|
assert len(SQLiteIdentityStore(store.database).list_accounts(limit=50, offset=0)) == 4
|
|
|
|
asyncio.run(exercise())
|
|
|
|
|
|
def test_password_login_origin_tls_size_and_throttle(tmp_path: Path) -> None:
|
|
async def exercise() -> None:
|
|
store = SQLiteIdentityStore(tmp_path / "identity.sqlite3")
|
|
store.create_account(
|
|
username="alice",
|
|
password=PASSWORD,
|
|
role="editor",
|
|
project_ids=("pilot",),
|
|
actor="operator",
|
|
)
|
|
app = create_app(
|
|
identity_provider=store,
|
|
account_identity=store,
|
|
node_reader=CanonicalNodeReader(GitCanonicalRepository(tmp_path / "knowledge.git")),
|
|
)
|
|
async with httpx2.AsyncClient(
|
|
transport=httpx2.ASGITransport(app=app), base_url="http://gewu.test"
|
|
) as http:
|
|
body = {"username": "alice", "password": PASSWORD}
|
|
assert (await http.post("/v1/users/session", json=body)).status_code == 403
|
|
assert (
|
|
await http.post(
|
|
"https://gewu.test/v1/users/session",
|
|
json=body,
|
|
headers={"Origin": "https://attacker.invalid"},
|
|
)
|
|
).status_code == 403
|
|
assert (
|
|
await http.post("https://gewu.test/v1/users/session", content=b" " * 4097)
|
|
).status_code == 413
|
|
for _ in range(5):
|
|
assert (
|
|
await http.post(
|
|
"https://gewu.test/v1/users/session", json={**body, "password": "wrong"}
|
|
)
|
|
).status_code == 401
|
|
assert (
|
|
await http.post("https://gewu.test/v1/users/session", json=body)
|
|
).status_code == 401
|
|
with store._connect() as db:
|
|
db.execute("UPDATE account_login_failures SET failed_at=0")
|
|
assert (
|
|
await http.post("https://gewu.test/v1/users/session", json=body)
|
|
).status_code == 200
|
|
|
|
asyncio.run(exercise())
|