"""Account identifiers, reset links, and address verification.""" from __future__ import annotations import os import re import sys import tempfile from pathlib import Path ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT)) os.environ["KANSHO_PROVIDER_KEY"] = "" os.environ["KANSHO_DETECT_PROVIDER_KEY"] = "" os.environ["KANSHO_FAKE_MAIL"] = "1" os.environ["KANSHO_ENV"] = "test" os.environ["APP_URL"] = "http://localhost:5188" if (os.environ.get("KANSHO_DB_BACKEND") or "").strip().lower() not in {"postgres", "postgresql", "pg"}: os.environ["KANSHO_DB_PATH"] = str(Path(tempfile.mkdtemp(prefix="kansho-account-")) / "account.sqlite") from fastapi.testclient import TestClient from main import app from db import get_db from mailer import OUTBOX, clear_outbox def expect(ok: bool, message: str) -> None: if not ok: raise SystemExit(f"FAIL: {message}") print(f"OK {message}") def auth_header(token: str) -> dict: return {"X-Auth-Token": token} def mail_tokens() -> list[str]: found: list[str] = [] for item in OUTBOX: found.extend(re.findall(r"token=([A-Za-z0-9_\-]+)", item["body"])) return found def main() -> None: clear_outbox() with TestClient(app) as client: setup = client.post( "/api/auth/setup", json={"email": "lars@example.test", "name": "Lars", "password": "test-pass"}, ) expect(setup.status_code == 200, f"setup {setup.text}") token = setup.json()["token"] headers = auth_header(token) verify_tokens = mail_tokens() expect(len(verify_tokens) == 1, "setup sends one verification link") verified = client.post("/api/auth/verify-email", json={"token": verify_tokens[0]}) expect(verified.status_code == 200 and verified.json()["email_verified"] is True, "admin email verified") clear_outbox() me = client.get("/api/auth/me", headers=headers) expect(me.status_code == 200, "session survives initial verification") admin_id = me.json()["id"] blocked = client.patch( f"/api/users/{admin_id}", headers=headers, json={"status": "disabled"}, ) expect(blocked.status_code == 400, f"last admin cannot be disabled {blocked.text}") created = client.post( "/api/users", headers=headers, json={ "email": "ute@example.test", "name": "Ute", "password": "user-pass", "role": "user", "username": "ute.user", }, ) expect(created.status_code == 200 and created.json()["username"] == "ute.user", f"create with username {created.text}") expect(created.json()["email_verified"] is False, "new primary email starts unverified") clear_outbox() by_name = client.post("/api/auth/login", json={"email": "Ute.User", "password": "user-pass"}) expect(by_name.status_code == 200, f"login by username {by_name.text}") unknown = client.post("/api/auth/forgot-password", json={"email": "missing@example.test"}) known = client.post("/api/auth/forgot-password", json={"email": "lars@example.test"}) expect(unknown.status_code == 200 and known.status_code == 200, "forgot password always 200") expect(unknown.json() == known.json(), "forgot password does not reveal whether the account exists") expect(len(mail_tokens()) == 1, "reset mail only for the verified account") reset_token = mail_tokens()[0] clear_outbox() changed = client.post( "/api/auth/reset-password", json={"token": reset_token, "password": "new-pass"}, ) expect(changed.status_code == 200, f"reset password {changed.text}") stale = client.get("/api/auth/me", headers=headers) expect(stale.status_code == 401, "reset deletes sessions") old_password = client.post("/api/auth/login", json={"email": "lars@example.test", "password": "test-pass"}) expect(old_password.status_code == 401, "old password rejected") relogin = client.post("/api/auth/login", json={"email": "lars@example.test", "password": "new-pass"}) expect(relogin.status_code == 200, "login with reset password") headers = auth_header(relogin.json()["token"]) again = client.post("/api/auth/reset-password", json={"token": reset_token, "password": "other-pass"}) expect(again.status_code == 400, "reset token is single use") client.post("/api/auth/forgot-password", json={"email": "lars@example.test"}) expired_token = mail_tokens()[0] clear_outbox() with get_db() as conn: conn.execute( "UPDATE password_reset_tokens SET expires_at = ? WHERE token_hash IS NOT NULL", ("2000-01-01 00:00:00",), ) expired = client.post( "/api/auth/reset-password", json={"token": expired_token, "password": "newer-pass"}, ) expect(expired.status_code == 400, "expired reset token is rejected") clear_outbox() renamed = client.patch( "/api/users/me", headers=headers, json={"email": "lars.new@example.test"}, ) expect(renamed.status_code == 200, f"email change queued {renamed.text}") expect(renamed.json()["email"] == "lars@example.test", "login email unchanged until verification") expect(renamed.json()["pending_email"] == "lars.new@example.test", "pending email stored") still = client.post("/api/auth/login", json={"email": "lars@example.test", "password": "new-pass"}) expect(still.status_code == 200, "old email still logs in") confirm = client.post("/api/auth/verify-email", json={"token": mail_tokens()[0]}) expect(confirm.status_code == 200 and confirm.json()["email"] == "lars.new@example.test", "verified address becomes login") gone = client.post("/api/auth/login", json={"email": "lars@example.test", "password": "new-pass"}) expect(gone.status_code == 401, "previous email stops working") moved = client.post("/api/auth/login", json={"email": "lars.new@example.test", "password": "new-pass"}) expect(moved.status_code == 200, "new email logs in") headers = auth_header(moved.json()["token"]) clear_outbox() other = client.post( f"/api/users/{created.json()['id']}/password-reset", headers=headers, ) expect(other.status_code == 200, f"admin reset for another user {other.text}") expect(len(mail_tokens()) == 1, "admin reset sends one link") self_reset = client.post(f"/api/users/{moved.json()['profile_id']}/password-reset", headers=headers) expect(self_reset.status_code == 400, "admin cannot reset the own account via the admin endpoint") os.environ.pop("KANSHO_FAKE_MAIL", None) os.environ.pop("SMTP_HOST", None) os.environ.pop("SMTP_FROM", None) os.environ["KANSHO_ENV"] = "test" failed = client.post( f"/api/users/{created.json()['id']}/password-reset", headers=headers, ) expect(failed.status_code == 503, f"admin reset without SMTP fails closed {failed.text}") secondary_login = client.post( "/api/auth/login", json={"email": "ute.second@example.test", "password": "user-pass"}, ) expect(secondary_login.status_code == 401, "unknown address is not a login") print("Account access tests passed.") if __name__ == "__main__": main()