310 lines
11 KiB
Python
310 lines
11 KiB
Python
 
|
"""Password reset by email, recovery address, and the system mailer (SPEC §6.1).
|
||
|
|
|
||
|
|
The SMTP send is replaced with a recorder, so nothing leaves the machine; the
|
||
|
|
reset link it would have sent is read back from what was recorded.
|
||
|
|
"""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import uuid
|
||
|
|
from collections.abc import AsyncIterator
|
||
|
|
from dataclasses import dataclass, field
|
||
|
|
|
||
|
|
import httpx
|
||
|
|
import pyotp
|
||
|
|
import pytest
|
||
|
|
import pytest_asyncio
|
||
|
|
from httpx import ASGITransport
|
||
|
|
from sqlalchemy import delete, update
|
||
|
|
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
|
||
|
|
from sqlalchemy.pool import NullPool
|
||
|
|
|
||
|
|
from app.api import admin as admin_api
|
||
|
|
from app.config import get_settings
|
||
|
|
from app.identity import routes
|
||
|
|
from app.identity.mailer import MailerConfig
|
||
|
|
from app.identity.passwords import hash_password
|
||
|
|
from app.identity.reset import MAX_TOTP_FAILURES
|
||
|
|
from app.identity.service import app_settings
|
||
|
|
from app.identity.sessions import create_session
|
||
|
|
from app.identity.totp import encrypt_totp_secret
|
||
|
|
from app.main import create_app
|
||
|
|
from app.models import AppSettings, User
|
||
|
|
from app.redis_client import get_redis
|
||
|
|
|
||
|
|
CSRF = "csrf-token"
|
||
|
|
PASSWORD = "hunter2pass"
|
||
|
|
NEW_PASSWORD = "brandNewPass9"
|
||
|
|
CFG = MailerConfig(
|
||
|
|
host="smtp.example.com",
|
||
|
|
port=587,
|
||
|
|
security="starttls",
|
||
|
|
username="postmaster@example.com",
|
||
|
|
password="pw",
|
||
|
|
sender="postmaster@example.com",
|
||
|
|
public_url="https://mail.example.com",
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
@dataclass
|
||
|
|
class Sent:
|
||
|
|
mails: list[tuple[str, str, str]] = field(default_factory=list) # (to, subject, text)
|
||
|
|
|
||
|
|
def token(self) -> str:
|
||
|
|
text = self.mails[-1][2]
|
||
|
|
link = next(w for w in text.split() if w.startswith("https://mail.example.com/"))
|
||
|
|
assert link.startswith("https://mail.example.com/reset-password#token=")
|
||
|
|
return link.split("#token=", 1)[1]
|
||
|
|
|
||
|
|
|
||
|
|
@dataclass
|
||
|
|
class Ctx:
|
||
|
|
session: AsyncSession
|
||
|
|
anon: httpx.AsyncClient
|
||
|
|
sent: Sent
|
||
|
|
created: list[uuid.UUID] = field(default_factory=list)
|
||
|
|
|
||
|
|
async def user(
|
||
|
|
self, *, recovery: str | None = "me@elsewhere.example", admin: bool = False
|
||
|
|
) -> User:
|
||
|
|
user = User(
|
||
|
|
handle=f"pr{uuid.uuid4().hex[:8]}@test.local",
|
||
|
|
password_hash=hash_password(PASSWORD),
|
||
|
|
recovery_email=recovery,
|
||
|
|
is_admin=admin,
|
||
|
|
)
|
||
|
|
self.session.add(user)
|
||
|
|
await self.session.commit()
|
||
|
|
self.created.append(user.id)
|
||
|
|
await get_redis().delete(f"pwreset-rate:{user.id}")
|
||
|
|
return user
|
||
|
|
|
||
|
|
async def signed_in(self, user: User) -> httpx.AsyncClient:
|
||
|
|
token = await create_session(user.id, "pytest")
|
||
|
|
return httpx.AsyncClient(
|
||
|
|
transport=ASGITransport(app=create_app()),
|
||
|
|
base_url="http://testserver",
|
||
|
|
cookies={"session": token, "csrf": CSRF},
|
||
|
|
headers={"X-CSRF": CSRF},
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
@pytest_asyncio.fixture
|
||
|
|
async def ctx(monkeypatch: pytest.MonkeyPatch) -> AsyncIterator[Ctx]:
|
||
|
|
sent = Sent()
|
||
|
|
|
||
|
|
async def fake_send(cfg: MailerConfig, to: str, subject: str, text: str, html: str) -> None:
|
||
|
|
sent.mails.append((to, subject, text))
|
||
|
|
|
||
|
|
monkeypatch.setattr(routes, "send_mail", fake_send)
|
||
|
|
monkeypatch.setattr(admin_api, "send_mail", fake_send)
|
||
|
|
monkeypatch.setattr(routes, "mailer_config", lambda row: CFG)
|
||
|
|
|
||
|
|
engine = create_async_engine(get_settings().database_url, poolclass=NullPool)
|
||
|
|
session = async_sessionmaker(engine, expire_on_commit=False)()
|
||
|
|
anon = httpx.AsyncClient(transport=ASGITransport(app=create_app()), base_url="http://t")
|
||
|
|
c = Ctx(session=session, anon=anon, sent=sent)
|
||
|
|
try:
|
||
|
|
yield c
|
||
|
|
finally:
|
||
|
|
await anon.aclose()
|
||
|
|
await session.execute(delete(User).where(User.id.in_(c.created)))
|
||
|
|
await session.commit()
|
||
|
|
await session.close()
|
||
|
|
await engine.dispose()
|
||
|
|
|
||
|
|
|
||
|
|
async def _request(ctx: Ctx, handle: str) -> httpx.Response:
|
||
|
|
return await ctx.anon.post("/api/auth/password/reset-request", json={"handle": handle})
|
||
|
|
|
||
|
|
|
||
|
|
async def _login(ctx: Ctx, handle: str, password: str) -> int:
|
||
|
|
resp = await ctx.anon.post("/api/auth/login", json={"handle": handle, "password": password})
|
||
|
|
return resp.status_code
|
||
|
|
|
||
|
|
|
||
|
|
# --- requesting a link -----------------------------------------------------------
|
||
|
|
|
||
|
|
|
||
|
|
async def test_request_mails_a_one_time_link_to_the_recovery_address(ctx: Ctx) -> None:
|
||
|
|
user = await ctx.user()
|
||
|
|
assert (await _request(ctx, user.handle)).status_code == 200
|
||
|
|
assert [m[0] for m in ctx.sent.mails] == ["me@elsewhere.example"]
|
||
|
|
token = ctx.sent.token()
|
||
|
|
|
||
|
|
check = await ctx.anon.post("/api/auth/password/reset/check", json={"token": token})
|
||
|
|
assert check.json() == {"valid": True, "totp_required": False}
|
||
|
|
|
||
|
|
reset = await ctx.anon.post(
|
||
|
|
"/api/auth/password/reset", json={"token": token, "password": NEW_PASSWORD}
|
||
|
|
)
|
||
|
|
assert reset.status_code == 200
|
||
|
|
# Used up: a second go fails, and the page is told so up front.
|
||
|
|
again = await ctx.anon.post(
|
||
|
|
"/api/auth/password/reset", json={"token": token, "password": "anotherPass12"}
|
||
|
|
)
|
||
|
|
assert again.json()["error"]["code"] == "invalid_token"
|
||
|
|
check = await ctx.anon.post("/api/auth/password/reset/check", json={"token": token})
|
||
|
|
assert check.json()["valid"] is False
|
||
|
|
|
||
|
|
assert await _login(ctx, user.handle, NEW_PASSWORD) == 200
|
||
|
|
assert await _login(ctx, user.handle, PASSWORD) == 401
|
||
|
|
|
||
|
|
|
||
|
|
async def test_the_answer_never_says_whether_the_account_exists(ctx: Ctx) -> None:
|
||
|
|
no_recovery = await ctx.user(recovery=None)
|
||
|
|
for handle in ("nobody-here-at-all", no_recovery.handle):
|
||
|
|
resp = await _request(ctx, handle)
|
||
|
|
assert resp.status_code == 200 and resp.json() == {"status": "ok"}
|
||
|
|
assert ctx.sent.mails == []
|
||
|
|
|
||
|
|
|
||
|
|
async def test_no_mail_while_system_mail_is_not_set_up(
|
||
|
|
ctx: Ctx, monkeypatch: pytest.MonkeyPatch
|
||
|
|
) -> None:
|
||
|
|
monkeypatch.setattr(routes, "mailer_config", lambda row: None)
|
||
|
|
user = await ctx.user()
|
||
|
|
assert (await _request(ctx, user.handle)).status_code == 200
|
||
|
|
assert ctx.sent.mails == []
|
||
|
|
|
||
|
|
|
||
|
|
async def test_requests_are_capped_per_hour(ctx: Ctx) -> None:
|
||
|
|
user = await ctx.user()
|
||
|
|
for _ in range(5):
|
||
|
|
assert (await _request(ctx, user.handle)).status_code == 200
|
||
|
|
assert len(ctx.sent.mails) == 3
|
||
|
|
|
||
|
|
|
||
|
|
# --- two-step verification ---------------------------------------------------------
|
||
|
|
|
||
|
|
|
||
|
|
async def test_2fa_code_is_required_before_the_password_changes(ctx: Ctx) -> None:
|
||
|
|
user = await ctx.user()
|
||
|
|
secret = pyotp.random_base32()
|
||
|
|
await ctx.session.execute(
|
||
|
|
update(User).where(User.id == user.id).values(totp_secret=encrypt_totp_secret(secret))
|
||
|
|
)
|
||
|
|
await ctx.session.commit()
|
||
|
|
await _request(ctx, user.handle)
|
||
|
|
token = ctx.sent.token()
|
||
|
|
body = {"token": token, "password": NEW_PASSWORD}
|
||
|
|
|
||
|
|
check = await ctx.anon.post("/api/auth/password/reset/check", json={"token": token})
|
||
|
|
assert check.json()["totp_required"] is True
|
||
|
|
missing = await ctx.anon.post("/api/auth/password/reset", json=body)
|
||
|
|
assert missing.json()["error"]["code"] == "totp_required"
|
||
|
|
wrong = await ctx.anon.post("/api/auth/password/reset", json={**body, "totp": "000000"})
|
||
|
|
assert wrong.json()["error"]["code"] == "totp_invalid"
|
||
|
|
# Nothing changed yet, and the link still works.
|
||
|
|
assert await _login(ctx, user.handle, PASSWORD) == 401 # 2FA account: needs a code
|
||
|
|
good = await ctx.anon.post(
|
||
|
|
"/api/auth/password/reset", json={**body, "totp": pyotp.TOTP(secret).now()}
|
||
|
|
)
|
||
|
|
assert good.status_code == 200
|
||
|
|
login = await ctx.anon.post(
|
||
|
|
"/api/auth/login",
|
||
|
|
json={"handle": user.handle, "password": NEW_PASSWORD, "totp": pyotp.TOTP(secret).now()},
|
||
|
|
)
|
||
|
|
assert login.status_code == 200
|
||
|
|
|
||
|
|
|
||
|
|
async def test_too_many_wrong_codes_burn_the_link(ctx: Ctx) -> None:
|
||
|
|
user = await ctx.user()
|
||
|
|
secret = pyotp.random_base32()
|
||
|
|
await ctx.session.execute(
|
||
|
|
update(User).where(User.id == user.id).values(totp_secret=encrypt_totp_secret(secret))
|
||
|
|
)
|
||
|
|
await ctx.session.commit()
|
||
|
|
await _request(ctx, user.handle)
|
||
|
|
token = ctx.sent.token()
|
||
|
|
body = {"token": token, "password": NEW_PASSWORD, "totp": "000000"}
|
||
|
|
for _ in range(MAX_TOTP_FAILURES):
|
||
|
|
await ctx.anon.post("/api/auth/password/reset", json=body)
|
||
|
|
late = await ctx.anon.post(
|
||
|
|
"/api/auth/password/reset", json={**body, "totp": pyotp.TOTP(secret).now()}
|
||
|
|
)
|
||
|
|
assert late.json()["error"]["code"] == "invalid_token"
|
||
|
|
|
||
|
|
|
||
|
|
# --- the recovery address ----------------------------------------------------------
|
||
|
|
|
||
|
|
|
||
|
|
async def test_changing_the_recovery_email_takes_the_password(ctx: Ctx) -> None:
|
||
|
|
user = await ctx.user(recovery=None)
|
||
|
|
async with await ctx.signed_in(user) as c:
|
||
|
|
wrong = await c.put(
|
||
|
|
"/api/auth/recovery-email",
|
||
|
|
json={"password": "not-it", "recovery_email": "new@example.com"},
|
||
|
|
)
|
||
|
|
assert wrong.status_code == 403
|
||
|
|
bad = await c.put(
|
||
|
|
"/api/auth/recovery-email", json={"password": PASSWORD, "recovery_email": "nope"}
|
||
|
|
)
|
||
|
|
assert bad.json()["error"]["code"] == "invalid_email"
|
||
|
|
ok = await c.put(
|
||
|
|
"/api/auth/recovery-email",
|
||
|
|
json={"password": PASSWORD, "recovery_email": " New@Example.com "},
|
||
|
|
)
|
||
|
|
assert ok.json()["recovery_email"] == "new@example.com"
|
||
|
|
removed = await c.put(
|
||
|
|
"/api/auth/recovery-email", json={"password": PASSWORD, "recovery_email": ""}
|
||
|
|
)
|
||
|
|
assert removed.json()["recovery_email"] is None
|
||
|
|
|
||
|
|
|
||
|
|
# --- admin: system mail settings -----------------------------------------------------
|
||
|
|
|
||
|
|
|
||
|
|
@pytest_asyncio.fixture
|
||
|
|
async def restore_settings(ctx: Ctx) -> AsyncIterator[None]:
|
||
|
|
row = await app_settings(ctx.session)
|
||
|
|
saved = {
|
||
|
|
k: getattr(row, k)
|
||
|
|
for k in (
|
||
|
|
"smtp_host",
|
||
|
|
"smtp_port",
|
||
|
|
"smtp_security",
|
||
|
|
"smtp_username",
|
||
|
|
"smtp_secret_ct",
|
||
|
|
"smtp_from",
|
||
|
|
"public_url",
|
||
|
|
)
|
||
|
|
}
|
||
|
|
await ctx.session.commit()
|
||
|
|
yield
|
||
|
|
await ctx.session.execute(update(AppSettings).values(**saved))
|
||
|
|
await ctx.session.commit()
|
||
|
|
|
||
|
|
|
||
|
|
async def test_admin_configures_system_mail_without_ever_seeing_the_password(
|
||
|
|
ctx: Ctx, restore_settings: None
|
||
|
|
) -> None:
|
||
|
|
admin = await ctx.user(admin=True)
|
||
|
|
async with await ctx.signed_in(admin) as c:
|
||
|
|
resp = await c.put(
|
||
|
|
"/api/admin/mail",
|
||
|
|
json={
|
||
|
|
"smtp_host": "smtp.example.com",
|
||
|
|
"smtp_port": 465,
|
||
|
|
"smtp_security": "ssl",
|
||
|
|
"smtp_username": "postmaster@example.com",
|
||
|
|
"smtp_password": "s3cret",
|
||
|
|
"smtp_from": "postmaster@example.com",
|
||
|
|
"public_url": "https://mail.example.com/",
|
||
|
|
},
|
||
|
|
)
|
||
|
|
out = resp.json()
|
||
|
|
assert out["has_password"] is True and out["configured"] is True
|
||
|
|
assert out["public_url"] == "https://mail.example.com"
|
||
|
|
assert "s3cret" not in resp.text
|
||
|
|
# Omitting the password keeps it; an empty one removes it.
|
||
|
|
kept = (await c.put("/api/admin/mail", json={"smtp_port": 587})).json()
|
||
|
|
assert kept["has_password"] is True
|
||
|
|
gone = (await c.put("/api/admin/mail", json={"smtp_password": ""})).json()
|
||
|
|
assert gone["has_password"] is False
|
||
|
|
bad = await c.put("/api/admin/mail", json={"public_url": "mail.example.com"})
|
||
|
|
assert bad.json()["error"]["code"] == "invalid_url"
|
||
|
|
test = await c.post("/api/admin/mail/test", json={"to": "me@example.com"})
|
||
|
|
assert test.status_code == 200
|
||
|
|
assert ctx.sent.mails[-1][:2] == ("me@example.com", "Ultimo Email test message")
|