From 2f2a6f3487df3b722225d36fed5d1e69266b16d6 Mon Sep 17 00:00:00 2001 From: Karolis Strazdas Date: Fri, 6 Nov 2026 22:16:03 +0200 Subject: [PATCH 1/2] Enforce verified account access and durable email verification --- .github/workflows/ci.yml | 14 + ...1e2f3456789_email_verification_evidence.py | 142 ++++++ .../dev-pages/src/pages/auth-sandbox/main.ts | 12 +- .../src/pages/powersync-sandbox/main.ts | 13 +- papyrus/api/deps.py | 26 +- papyrus/api/routes/auth.py | 45 +- papyrus/config.py | 2 +- papyrus/core/database.py | 1 + papyrus/core/exceptions.py | 28 ++ papyrus/main.py | 46 +- papyrus/models/__init__.py | 4 + papyrus/models/auth.py | 30 ++ papyrus/models/user.py | 14 +- papyrus/schemas/auth.py | 12 + papyrus/services/account_deletion.py | 5 + papyrus/services/auth/_core.py | 44 +- papyrus/services/auth/email_flows.py | 62 ++- papyrus/services/auth/google.py | 148 ++++-- papyrus/services/auth/sessions.py | 99 ++-- papyrus/services/auth/types.py | 1 + papyrus/services/auth/verification.py | 226 +++++++++ tests/api/routes/test_auth.py | 66 +-- tests/api/routes/test_avatars.py | 22 +- tests/api/routes/test_rate_limits.py | 3 + tests/api/routes/test_verification.py | 184 ++++++++ tests/conftest.py | 3 + tests/services/test_auth.py | 53 ++- .../test_auth_recovery_concurrency.py | 14 +- tests/services/test_verification.py | 439 ++++++++++++++++++ tests/test_models.py | 4 + tests/test_openapi_export.py | 3 + tests/test_verification_migration.py | 126 +++++ 32 files changed, 1677 insertions(+), 214 deletions(-) create mode 100644 alembic/versions/d1e2f3456789_email_verification_evidence.py create mode 100644 papyrus/services/auth/verification.py create mode 100644 tests/api/routes/test_verification.py create mode 100644 tests/services/test_verification.py create mode 100644 tests/test_verification_migration.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 17a98ac..957dd39 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -68,6 +68,20 @@ jobs: - name: Mypy run: uv run mypy . + - name: Export public API snapshot + run: | + mkdir -p build/api + uv run --locked python scripts/export_openapi.py build/api/openapi.json + uv run --locked python scripts/export_openapi.py build/api/openapi.json --check + + - name: Upload public API snapshot + uses: actions/upload-artifact@v4 + with: + name: server-openapi-${{ github.sha }} + path: build/api/openapi.json + if-no-files-found: error + retention-days: 14 + - name: Pytest env: PAPYRUS_TEST_TIMING: build/test-results/timing.json diff --git a/alembic/versions/d1e2f3456789_email_verification_evidence.py b/alembic/versions/d1e2f3456789_email_verification_evidence.py new file mode 100644 index 0000000..68f97dc --- /dev/null +++ b/alembic/versions/d1e2f3456789_email_verification_evidence.py @@ -0,0 +1,142 @@ +"""Bind verification to email and retain only attributable legacy evidence. + +Password-only verified accounts retain Papyrus evidence. Matching retained Google +Gmail identities retain Google evidence. Other unattributable booleans need proof. No users, +credentials or library data are deleted. The activation audit reports the count +of previously verified accounts that now need proof. Rollback must fix forward. +""" + +import logging +from collections.abc import Sequence + +import sqlalchemy as sa + +from alembic import op + +revision: str = "d1e2f3456789" +down_revision: str | Sequence[str] | None = "c0d1e2f34567" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + +TRUSTED_PASSWORD = """ + primary_email_verified = true AND primary_email IS NOT NULL + AND EXISTS (SELECT 1 FROM password_credentials p WHERE p.user_id = users.user_id) + AND NOT EXISTS (SELECT 1 FROM user_identities i WHERE i.user_id = users.user_id) +""" + + +TRUSTED_GOOGLE = """ + primary_email_verified = true AND primary_email IS NOT NULL + AND EXISTS ( + SELECT 1 FROM user_identities i WHERE i.user_id = users.user_id + AND i.provider = 'google' AND i.provider_subject <> '' + AND i.email_at_provider = users.primary_email + AND i.email_at_provider LIKE '%@gmail.com' + ) +""" + +TRUSTED_LEGACY = f"({TRUSTED_PASSWORD}) OR ({TRUSTED_GOOGLE})" + + +def upgrade() -> None: + for name, kind in ( + ("verified_email", sa.String(255)), + ("verification_source", sa.String(32)), + ("verification_subject", sa.String(255)), + ("verified_at", sa.DateTime(timezone=True)), + ): + op.add_column("users", sa.Column(name, kind, nullable=True)) + + for table in ("user_identities", "auth_exchange_codes"): + op.add_column(table, sa.Column("hosted_domain", sa.String(255), nullable=True)) + + op.add_column("email_action_tokens", sa.Column("email", sa.String(255), nullable=True)) + + affected = op.get_bind().scalar( + sa.text(f"SELECT count(*) FROM users WHERE primary_email_verified = true AND NOT ({TRUSTED_LEGACY})"), + ) + + logging.getLogger("alembic.runtime.migration").info( + "Verification evidence audit: %s accounts require fresh proof", affected + ) + + op.execute( + sa.text( + f"UPDATE users SET verified_email = primary_email, verification_source = 'papyrus_email', " + f"verified_at = CURRENT_TIMESTAMP WHERE {TRUSTED_PASSWORD}", + ), + ) + + op.execute( + sa.text( + f""" + UPDATE users SET verified_email = primary_email, verification_source = 'google', + verification_subject = ( + SELECT i.provider_subject FROM user_identities i + WHERE i.user_id = users.user_id AND i.provider = 'google' + AND i.email_at_provider = users.primary_email AND i.provider_subject <> '' + ORDER BY i.created_at, i.identity_id LIMIT 1 + ), + verified_at = CURRENT_TIMESTAMP + WHERE {TRUSTED_GOOGLE} + """, + ), + ) + + op.execute("UPDATE users SET primary_email_verified = false WHERE verified_email IS NULL") + op.execute("UPDATE email_action_tokens SET used_at = CURRENT_TIMESTAMP WHERE action_type = 'verify_email'") + + op.execute( + "UPDATE auth_sessions SET revoked_at = CURRENT_TIMESTAMP WHERE revoked_at IS NULL " + "AND user_id IN (SELECT user_id FROM users WHERE verified_email IS NULL)", + ) + + op.create_table( + "verification_recipients", + sa.Column("email", sa.String(255), primary_key=True), + sa.Column("last_queued_at", sa.DateTime(timezone=True), nullable=False), + ) + + op.create_table( + "verification_deliveries", + sa.Column( + "token_id", sa.Uuid(), sa.ForeignKey("email_action_tokens.token_id", ondelete="CASCADE"), primary_key=True + ), + sa.Column("user_id", sa.Uuid(), sa.ForeignKey("users.user_id", ondelete="CASCADE"), nullable=False), + sa.Column("recipient", sa.String(255), nullable=False), + sa.Column("token", sa.Text(), nullable=False), + sa.Column("attempts", sa.Integer(), server_default=sa.text("0"), nullable=False), + sa.Column("next_attempt_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("expires_at", sa.DateTime(timezone=True), nullable=False), + ) + + op.create_index("ix_verification_deliveries_user_id", "verification_deliveries", ["user_id"]) + op.create_index("ix_verification_deliveries_next_attempt_at", "verification_deliveries", ["next_attempt_at"]) + + op.execute( + """ + DO $$ + DECLARE permission record; + BEGIN + REVOKE ALL ON TABLE verification_deliveries, verification_recipients FROM PUBLIC; + FOR permission IN + SELECT DISTINCT relation.relname, role.rolname + FROM pg_class relation + JOIN pg_namespace namespace ON namespace.oid = relation.relnamespace + CROSS JOIN LATERAL aclexplode(relation.relacl) acl + JOIN pg_roles role ON role.oid = acl.grantee + WHERE namespace.nspname = current_schema() + AND relation.relname IN ('verification_deliveries', 'verification_recipients') + AND acl.grantee <> relation.relowner + LOOP + EXECUTE format('REVOKE ALL ON TABLE %I FROM %I CASCADE', permission.relname, permission.rolname); + END LOOP; + END $$ + """, + ) + + +def downgrade() -> None: + raise RuntimeError( + "Verification enforcement must be retained. Fix forward without removing evidence or pending accounts." + ) diff --git a/frontend/dev-pages/src/pages/auth-sandbox/main.ts b/frontend/dev-pages/src/pages/auth-sandbox/main.ts index 5690b60..d56b41d 100644 --- a/frontend/dev-pages/src/pages/auth-sandbox/main.ts +++ b/frontend/dev-pages/src/pages/auth-sandbox/main.ts @@ -34,6 +34,13 @@ interface CurrentUser { display_name?: string; } +interface PendingVerificationPayload { + status: "verification_required"; + email: string; + message: string; + resend_after: number; +} + interface AuthTokensPayload { access_token?: string; refresh_token?: string; @@ -349,7 +356,7 @@ async function handleOAuthReturn(): Promise { function registerHandlers(): void { refs.register.onclick = async () => { await runButtonAction("register", refs.register, "Registering", "Registering user", async () => { - const { response, body } = await callSandboxApi(config.registerUrl, { + const { response, body } = await callSandboxApi(config.registerUrl, { method: "POST", body: JSON.stringify({ email: refs.email.value, @@ -361,8 +368,7 @@ function registerHandlers(): void { }); if (response.ok && body && typeof body === "object") { - applyTokens(body as AuthTokensPayload); - markOutcome("success", "Registration succeeded and tokens were stored."); + markOutcome("success", "Check your email to verify your account, then sign in."); return; } diff --git a/frontend/dev-pages/src/pages/powersync-sandbox/main.ts b/frontend/dev-pages/src/pages/powersync-sandbox/main.ts index 62d5995..71ee118 100644 --- a/frontend/dev-pages/src/pages/powersync-sandbox/main.ts +++ b/frontend/dev-pages/src/pages/powersync-sandbox/main.ts @@ -32,6 +32,13 @@ interface CurrentUser { email?: string | null; } +interface PendingVerificationPayload { + status: "verification_required"; + email: string; + message: string; + resend_after: number; +} + interface AuthTokensPayload { access_token?: string; refresh_token?: string; @@ -549,7 +556,7 @@ async function handleOAuthReturn(): Promise { function registerHandlers(): void { refs.register.onclick = async () => { await runButtonAction("register", refs.register, "Registering", "Registering user", async () => { - const { response, body } = await callSandboxApi(config.register_url, { + const { response, body } = await callSandboxApi(config.register_url, { method: "POST", body: JSON.stringify({ email: refs.email.value, @@ -561,9 +568,7 @@ function registerHandlers(): void { }); if (response.ok && body && typeof body === "object") { - applyTokens(body as AuthTokensPayload); - await refreshServerItems(); - markOutcome("success", "Registration succeeded."); + markOutcome("success", "Check your email to verify your account, then sign in."); return; } diff --git a/papyrus/api/deps.py b/papyrus/api/deps.py index b86893d..89259ca 100644 --- a/papyrus/api/deps.py +++ b/papyrus/api/deps.py @@ -12,7 +12,7 @@ from papyrus.core.database import get_db from papyrus.core.security import decode_token -from papyrus.models import AuthSession +from papyrus.models import AuthSession, User security = HTTPBearer() @@ -39,7 +39,7 @@ async def get_current_access_token_payload( return payload -async def get_current_auth_session( +async def get_revocable_auth_session( payload: Annotated[dict[str, Any], Depends(get_current_access_token_payload)], db: Annotated[AsyncSession, Depends(get_db)], ) -> AuthSession: @@ -62,6 +62,13 @@ async def get_current_auth_session( detail={"code": "INVALID_TOKEN", "message": "Token contains malformed identifiers"}, ) from exc + await db.scalar( + select(User) + .where(User.user_id == user_uuid) + .with_for_update(key_share=True) + .execution_options(populate_existing=True), + ) + result = await db.execute( select(AuthSession).options(selectinload(AuthSession.user)).where(AuthSession.session_id == session_uuid) ) @@ -80,7 +87,7 @@ async def get_current_auth_session( detail={"code": "SESSION_REVOKED", "message": "Session is expired or revoked"}, ) - if auth_session.user.disabled_at is not None: + if auth_session.user.disabled_at is not None or auth_session.user.deletion_due_at is not None: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail={"code": "ACCOUNT_DISABLED", "message": "User account is disabled"}, @@ -89,6 +96,19 @@ async def get_current_auth_session( return auth_session +async def get_current_auth_session( + auth_session: Annotated[AuthSession, Depends(get_revocable_auth_session)], + db: Annotated[AsyncSession, Depends(get_db)], +) -> AuthSession: + from papyrus.services.auth.verification import require_verified + + await require_verified(db, auth_session.user) + return auth_session + + +RevocableAuthSession = Annotated[AuthSession, Depends(get_revocable_auth_session)] + + async def get_current_user_id( request: Request, auth_session: Annotated[AuthSession, Depends(get_current_auth_session)], diff --git a/papyrus/api/routes/auth.py b/papyrus/api/routes/auth.py index e601840..83e4980 100644 --- a/papyrus/api/routes/auth.py +++ b/papyrus/api/routes/auth.py @@ -9,7 +9,7 @@ from fastapi.responses import RedirectResponse from sqlalchemy.ext.asyncio import AsyncSession -from papyrus.api.deps import CurrentSessionId, CurrentUserId +from papyrus.api.deps import CurrentUserId, RevocableAuthSession from papyrus.config import get_settings from papyrus.core.database import get_db from papyrus.core.rate_limit import limiter @@ -22,24 +22,37 @@ OAuthExchangeRequest, OAuthLinkCompleteRequest, OAuthStartRequest, + PendingVerificationResponse, PowerSyncTokenResponse, RefreshTokenRequest, RegisterRequest, ResendVerificationRequest, + ResendVerificationResponse, ResetPasswordRequest, VerifyEmailRequest, ) -from papyrus.schemas.common import MessageResponse +from papyrus.schemas.common import Error, MessageResponse from papyrus.schemas.user import User from papyrus.services import auth as auth_service from papyrus.services.users import user_response -router = APIRouter() +router = APIRouter( + responses={ + 400: {"model": Error, "description": "Invalid request or VERIFICATION_TOKEN_INVALID."}, + 401: {"model": Error, "description": "Invalid, expired or revoked credentials."}, + 403: {"model": Error, "description": "EMAIL_VERIFICATION_REQUIRED includes email and resend_after seconds."}, + 429: {"model": Error, "description": "RATE_LIMIT_EXCEEDED with Retry-After and details.retry_after seconds."}, + 503: { + "model": Error, + "description": "VERIFICATION_UNAVAILABLE with Retry-After: 60. Retry sign-in or resend after an uncertain registration outcome.", + }, + }, +) DBSession = Annotated[AsyncSession, Depends(get_db)] def _auth_rate_limit() -> str: - return f"{get_settings().rate_limit_auth}/minute" + return f"{min(5, get_settings().rate_limit_auth)}/minute" async def _auth_tokens_response(db: AsyncSession, result: auth_service.AuthResult) -> AuthTokens: @@ -75,7 +88,7 @@ def _public_callback_url(request: Request, route_name: str) -> str: @router.post( "/register", - response_model=AuthTokens, + response_model=PendingVerificationResponse, status_code=status.HTTP_201_CREATED, summary="Register a new user account", ) @@ -84,10 +97,9 @@ async def register_user( request: Request, payload: RegisterRequest, db: DBSession, -) -> AuthTokens: +) -> PendingVerificationResponse: """Create a new user account with email and password.""" - result = await auth_service.register_user(db, payload, request.headers.get("user-agent")) - return await _auth_tokens_response(db, result) + return await auth_service.register_user(db, payload, request.headers.get("user-agent")) @router.post( @@ -187,16 +199,15 @@ async def refresh_token( summary="Logout and invalidate the current session", ) async def logout_user( - user_id: CurrentUserId, - session_id: CurrentSessionId, + auth_session: RevocableAuthSession, db: DBSession, request: LogoutRequest | None = None, ) -> Response: """Invalidate the current session or all sessions if requested.""" if request is not None and request.all_devices: - await auth_service.logout_all_sessions(db, user_id) + await auth_service.logout_all_sessions(db, auth_session.user_id) else: - await auth_service.logout_current_session(db, user_id, session_id) + await auth_service.logout_current_session(db, auth_session.user_id, auth_session.session_id) return Response(status_code=status.HTTP_204_NO_CONTENT) @@ -207,11 +218,11 @@ async def logout_user( summary="Logout from all sessions", ) async def logout_all( - user_id: CurrentUserId, + auth_session: RevocableAuthSession, db: DBSession, ) -> Response: """Invalidate all refresh-token backed sessions for the user.""" - await auth_service.logout_all_sessions(db, user_id) + await auth_service.logout_all_sessions(db, auth_session.user_id) return Response(status_code=status.HTTP_204_NO_CONTENT) @@ -288,7 +299,7 @@ async def verify_email( @router.post( "/resend-verification", - response_model=MessageResponse, + response_model=ResendVerificationResponse, summary="Resend verification email", ) @limiter.limit(_auth_rate_limit) @@ -296,10 +307,10 @@ async def resend_verification( request: Request, payload: ResendVerificationRequest, db: DBSession, -) -> MessageResponse: +) -> ResendVerificationResponse: """Issue a new verification token when email delivery is enabled.""" message = await auth_service.resend_verification_email(db, payload.email) - return MessageResponse(message=message) + return ResendVerificationResponse(message=message) @router.post( diff --git a/papyrus/config.py b/papyrus/config.py index 0507ea4..df9af21 100644 --- a/papyrus/config.py +++ b/papyrus/config.py @@ -62,7 +62,7 @@ class Settings(BaseSettings): oauth_allowed_redirect_hosts: list[str] = [] oauth_state_expire_minutes: int = 10 auth_exchange_code_expire_minutes: int = 5 - email_verification_token_expire_minutes: int = 1440 + email_verification_token_expire_minutes: int = Field(default=1440, ge=1440, le=1440) password_reset_token_expire_minutes: int = 60 email_delivery_enabled: bool = False smtp_host: str | None = None diff --git a/papyrus/core/database.py b/papyrus/core/database.py index eab01a7..bedd785 100644 --- a/papyrus/core/database.py +++ b/papyrus/core/database.py @@ -19,6 +19,7 @@ class Base(DeclarativeBase): engine = create_async_engine( settings.database_url, echo=settings.debug, + hide_parameters=True, pool_pre_ping=True, ) diff --git a/papyrus/core/exceptions.py b/papyrus/core/exceptions.py index 13dd415..95d09ba 100644 --- a/papyrus/core/exceptions.py +++ b/papyrus/core/exceptions.py @@ -114,3 +114,31 @@ def __init__( status_code=429, details=details, ) + + +class VerificationRequiredError(AppError): + def __init__(self, email: str | None, resend_after: int) -> None: + super().__init__( + "Verify your email address before signing in.", + code="EMAIL_VERIFICATION_REQUIRED", + status_code=403, + details={"email": email, "resend_after": resend_after}, + ) + + +class VerificationUnavailableError(AppError): + def __init__(self) -> None: + super().__init__( + "Verification email is temporarily unavailable. Please try again later.", + code="VERIFICATION_UNAVAILABLE", + status_code=503, + ) + + +class VerificationTokenInvalidError(ValidationError): + def __init__(self) -> None: + super().__init__( + "This verification link is invalid or expired. Request a new link or sign in if you already verified.", + ) + + self.code = "VERIFICATION_TOKEN_INVALID" diff --git a/papyrus/main.py b/papyrus/main.py index d27b21b..f973558 100644 --- a/papyrus/main.py +++ b/papyrus/main.py @@ -14,6 +14,8 @@ from fastapi.responses import JSONResponse from fastapi.staticfiles import StaticFiles from slowapi.errors import RateLimitExceeded +from starlette.middleware.base import RequestResponseEndpoint +from starlette.responses import Response from papyrus import __version__ from papyrus.api.routes import api_router, include_debug_routers @@ -23,11 +25,32 @@ from papyrus.core.exceptions import AppError from papyrus.core.rate_limit import limiter, rate_limit_exceeded_handler from papyrus.services import account_deletion, acquisition_monitor +from papyrus.services.auth import verification settings = get_settings() +class AuthQueryFilter(logging.Filter): + """Keep OAuth state, codes and any token query parameters out of access logs.""" + + def filter(self, record: logging.LogRecord) -> bool: + arguments = record.args + + if isinstance(arguments, tuple) and len(arguments) == 5: + client, method, path, version, status_code = arguments + + if isinstance(path, str) and ("/auth/" in path or "/account-deletion" in path): + record.args = (client, method, path.split("?", 1)[0], version, status_code) + + return True + + def configure_logging() -> None: + access_logger = logging.getLogger("uvicorn.access") + + if not any(isinstance(item, AuthQueryFilter) for item in access_logger.filters): + access_logger.addFilter(AuthQueryFilter()) + structlog.configure( processors=[ structlog.contextvars.merge_contextvars, @@ -47,6 +70,9 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]: cleanup_task = asyncio.create_task( account_deletion.run_cleanup(async_session_maker), name="account-deletion-cleanup" ) + verification_task = asyncio.create_task( + verification.run_delivery(async_session_maker), name="verification-delivery" + ) monitor_task: asyncio.Task[None] | None = None if settings.acquisition_enabled and settings.acquisition_import_root is not None: @@ -61,6 +87,11 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]: try: yield finally: + verification_task.cancel() + + with suppress(asyncio.CancelledError): + await verification_task + cleanup_task.cancel() with suppress(asyncio.CancelledError): @@ -102,7 +133,7 @@ def create_app() -> FastAPI: ## Authentication Most endpoints require authentication via JWT Bearer token. Obtain tokens -through the `{auth_base_path}/register`, `{auth_base_path}/login`, or the +after email verification through `{auth_base_path}/login` or the browser-based `{auth_base_path}/oauth/google/start` flow. ``` @@ -153,10 +184,22 @@ def create_app() -> FastAPI: expose_headers=["X-OPDS-URL", "Retry-After"], ) + @app.middleware("http") + async def no_store_auth(request: Request, call_next: RequestResponseEndpoint) -> Response: + response = await call_next(request) + + if request.url.path.startswith( + (auth_base_path + "/", f"{settings.api_prefix}/account-deletion") + ) or response.status_code in {401, 403}: + response.headers["Cache-Control"] = "no-store" + + return response + @app.exception_handler(AppError) async def app_exception_handler(request: Request, exc: AppError) -> JSONResponse: return JSONResponse( status_code=exc.status_code, + headers={"Retry-After": "60"} if exc.code == "VERIFICATION_UNAVAILABLE" else None, content={ "error": { "code": exc.code, @@ -173,6 +216,7 @@ async def general_exception_handler(request: Request, exc: Exception) -> JSONRes return JSONResponse( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + headers={"Cache-Control": "no-store"}, content={ "error": { "code": "INTERNAL_ERROR", diff --git a/papyrus/models/__init__.py b/papyrus/models/__init__.py index f492a86..cce0816 100644 --- a/papyrus/models/__init__.py +++ b/papyrus/models/__init__.py @@ -7,6 +7,8 @@ EmailActionToken, PasswordCredential, UserIdentity, + VerificationDelivery, + VerificationRecipient, ) from papyrus.models.library import ( SyncAnnotation, @@ -53,4 +55,6 @@ "SyncBook", "User", "UserIdentity", + "VerificationDelivery", + "VerificationRecipient", ] diff --git a/papyrus/models/auth.py b/papyrus/models/auth.py index 10b303d..37184ac 100644 --- a/papyrus/models/auth.py +++ b/papyrus/models/auth.py @@ -22,6 +22,7 @@ class UserIdentity(Base): provider: Mapped[str] = mapped_column(String(32), nullable=False) provider_subject: Mapped[str] = mapped_column(String(255), nullable=False) email_at_provider: Mapped[str | None] = mapped_column(String(255), nullable=True) + hosted_domain: Mapped[str | None] = mapped_column(String(255), nullable=True) email_verified_at_provider: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False, server_default=func.now()) last_used_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) @@ -93,6 +94,7 @@ class AuthExchangeCode(Base): provider: Mapped[str | None] = mapped_column(String(32), nullable=True) provider_subject: Mapped[str | None] = mapped_column(String(255), nullable=True) email_at_provider: Mapped[str | None] = mapped_column(String(255), nullable=True) + hosted_domain: Mapped[str | None] = mapped_column(String(255), nullable=True) email_verified_at_provider: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) display_name: Mapped[str | None] = mapped_column(String(255), nullable=True) avatar_url: Mapped[str | None] = mapped_column(String(2048), nullable=True) @@ -110,6 +112,7 @@ class EmailActionToken(Base): token_id: Mapped[UUID] = mapped_column(Uuid, primary_key=True, default=uuid4) user_id: Mapped[UUID] = mapped_column(ForeignKey("users.user_id", ondelete="CASCADE"), nullable=False, index=True) + email: Mapped[str | None] = mapped_column(String(255), nullable=True) action_type: Mapped[str] = mapped_column(String(32), nullable=False) token_hash: Mapped[str] = mapped_column(String(64), nullable=False, unique=True) expires_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False) @@ -135,3 +138,30 @@ class AccountNotification(Base): created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False, server_default=func.now()) next_attempt_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False, index=True) expires_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False) + + +class VerificationRecipient(Base): + """Persist the normalized recipient cooldown across workers and account transitions.""" + + __tablename__ = "verification_recipients" + + email: Mapped[str] = mapped_column(String(255), primary_key=True) + last_queued_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False) + + +class VerificationDelivery(Base): + """Restricted outbox containing deliverable secrets only until delivery or invalidation.""" + + __tablename__ = "verification_deliveries" + + token_id: Mapped[UUID] = mapped_column( + ForeignKey("email_action_tokens.token_id", ondelete="CASCADE"), + primary_key=True, + ) + + user_id: Mapped[UUID] = mapped_column(ForeignKey("users.user_id", ondelete="CASCADE"), nullable=False, index=True) + recipient: Mapped[str] = mapped_column(String(255), nullable=False) + token: Mapped[str] = mapped_column(Text, nullable=False) + attempts: Mapped[int] = mapped_column(Integer, nullable=False, default=0, server_default=text("0")) + next_attempt_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False, index=True) + expires_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False) diff --git a/papyrus/models/user.py b/papyrus/models/user.py index 18d14e7..23d84f9 100644 --- a/papyrus/models/user.py +++ b/papyrus/models/user.py @@ -28,6 +28,11 @@ class User(Base): server_default=text("false"), ) + verified_email: Mapped[str | None] = mapped_column(String(255), nullable=True) + verification_source: Mapped[str | None] = mapped_column(String(32), nullable=True) + verification_subject: Mapped[str | None] = mapped_column(String(255), nullable=True) + verified_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False, server_default=func.now()) last_login_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) disabled_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) @@ -57,7 +62,14 @@ def email(self) -> str | None: @property def email_verified(self) -> bool: - return self.primary_email_verified + return ( + self.primary_email_verified + and self.primary_email is not None + and self.verified_email == self.primary_email + and self.verification_source in {"papyrus_email", "google"} + and self.verified_at is not None + and (self.verification_source != "google" or self.verification_subject is not None) + ) from papyrus.models.auth import AuthSession, PasswordCredential, UserIdentity # noqa: E402 diff --git a/papyrus/schemas/auth.py b/papyrus/schemas/auth.py index e9f091d..0226f4d 100644 --- a/papyrus/schemas/auth.py +++ b/papyrus/schemas/auth.py @@ -24,6 +24,18 @@ class RegisterRequest(BaseModel): device_label: str | None = Field(default=None, max_length=255) +class PendingVerificationResponse(BaseModel): + status: Literal["verification_required"] = "verification_required" + email: EmailStr + message: str = "Check your email to verify your account, then sign in." + resend_after: int = Field(default=60, ge=0) + + +class ResendVerificationResponse(BaseModel): + message: str = "If the email is registered, a verification link has been sent" + resend_after: int = Field(default=60, ge=0) + + class LoginRequest(BaseModel): """User login request.""" diff --git a/papyrus/services/account_deletion.py b/papyrus/services/account_deletion.py index 945ad50..250bd38 100644 --- a/papyrus/services/account_deletion.py +++ b/papyrus/services/account_deletion.py @@ -22,6 +22,7 @@ EmailActionToken, PasswordCredential, User, + VerificationDelivery, ) from papyrus.schemas.account_deletion import AccountDeletionAuthorization, AccountDeletionStatus, AccountPasswordRequest from papyrus.services.account_notifications import deliver_pending_notifications, queue_account_notification @@ -166,6 +167,7 @@ async def exchange_google(session: AsyncSession, code: str, verifier: str) -> Ac async def _revoke_credentials(session: AsyncSession, user_id: UUID) -> None: now = _now() + await session.execute(delete(VerificationDelivery).where(VerificationDelivery.user_id == user_id)) await session.execute(update(AuthSession).where(AuthSession.user_id == user_id).values(revoked_at=now)) await session.execute(update(AuthExchangeCode).where(AuthExchangeCode.user_id == user_id).values(used_at=now)) await session.execute(update(EmailActionToken).where(EmailActionToken.user_id == user_id).values(used_at=now)) @@ -228,6 +230,9 @@ async def cancel_deletion( if user.deletion_due_at is None: raise ValidationError("No account deletion is scheduled") + from papyrus.services.auth.verification import require_verified + + await require_verified(session, user) user.deletion_requested_at = None user.deletion_due_at = None user.disabled_at = None diff --git a/papyrus/services/auth/_core.py b/papyrus/services/auth/_core.py index ebf86df..447e5f1 100644 --- a/papyrus/services/auth/_core.py +++ b/papyrus/services/auth/_core.py @@ -57,6 +57,13 @@ async def _create_session_for_user( device_label: str | None, user_agent: str | None, ) -> AuthResult: + from papyrus.services.auth.verification import require_verified + + await require_verified(session, user) + + if user.disabled_at is not None or user.deletion_due_at is not None: + raise ForbiddenError("User account is disabled") + refresh_token = generate_opaque_token() refresh_token_hash = hash_opaque_token(refresh_token) normalized_client_type, normalized_device_label = _get_client_metadata(client_type, device_label, user_agent) @@ -118,6 +125,7 @@ async def _create_exchange_code( provider: str | None = None, provider_subject: str | None = None, email_at_provider: str | None = None, + hosted_domain: str | None = None, email_verified_at_provider: datetime | None = None, display_name: str | None = None, avatar_url: str | None = None, @@ -133,6 +141,7 @@ async def _create_exchange_code( provider=provider, provider_subject=provider_subject, email_at_provider=email_at_provider, + hosted_domain=hosted_domain, email_verified_at_provider=email_verified_at_provider, display_name=display_name, avatar_url=avatar_url, @@ -156,10 +165,10 @@ async def _lock_credential_owner(session: AsyncSession, user_id: UUID | None) -> ) -async def _consume_exchange_code(session: AsyncSession, code: str, purpose: str) -> AuthExchangeCode: +async def _consume_exchange_code(session: AsyncSession, code: str, purpose: str | tuple[str, ...]) -> AuthExchangeCode: query = select(AuthExchangeCode).where( AuthExchangeCode.code_hash == hash_opaque_token(code), - AuthExchangeCode.purpose == purpose, + AuthExchangeCode.purpose.in_((purpose,) if isinstance(purpose, str) else purpose), AuthExchangeCode.used_at.is_(None), ) @@ -227,17 +236,6 @@ def _default_display_name(identity: GoogleIdentity | None = None, email: str | N return "Papyrus User" -def _build_api_url(path: str) -> str | None: - public_base_url = get_settings().public_base_url - - if public_base_url is None: - return None - - base = public_base_url.rstrip("/") - prefix = get_settings().api_prefix - return f"{base}{prefix}{path}" - - def _build_app_url(path: str, query: dict[str, str] | None = None) -> str | None: app_public_base_url = get_settings().app_public_base_url @@ -254,26 +252,6 @@ def _build_app_url(path: str, query: dict[str, str] | None = None) -> str | None return url -def _verification_email_body(token: str) -> str: - verify_url = _build_api_url("/auth/verify-email") - - lines = [ - "Verify your Papyrus email address.", - "", - f"Verification token: {token}", - ] - - if verify_url is not None: - lines.extend( - [ - "", - f"Submit this token to: {verify_url}", - ] - ) - - return "\n".join(lines) - - def _password_reset_email_body(token: str) -> str: settings = get_settings() reset_url = _build_app_url("/reset-password", {"token": token}) diff --git a/papyrus/services/auth/email_flows.py b/papyrus/services/auth/email_flows.py index b809f39..3df80f6 100644 --- a/papyrus/services/auth/email_flows.py +++ b/papyrus/services/auth/email_flows.py @@ -3,15 +3,16 @@ from __future__ import annotations from sqlalchemy import select +from sqlalchemy.exc import SQLAlchemyError from sqlalchemy.ext.asyncio import AsyncSession from papyrus.config import get_settings +from papyrus.core.exceptions import ValidationError, VerificationTokenInvalidError, VerificationUnavailableError from papyrus.core.security import hash_password -from papyrus.models import PasswordCredential +from papyrus.models import PasswordCredential, User from papyrus.services import email as email_service from papyrus.services.auth._core import ( _consume_email_action_token, - _get_active_user, _get_user_by_email, _issue_email_action_token, _normalize_email, @@ -19,42 +20,53 @@ _password_reset_email_body, _password_reset_email_html, _revoke_user_sessions, - _verification_email_body, ) from papyrus.services.auth.types import EMAIL_VERIFICATION_ACTION, PASSWORD_RESET_ACTION +from papyrus.services.auth.verification import ( + accepts_verification, + cancel_verification_deliveries, + queue_verification, + record_verification, + require_delivery_configuration, +) async def resend_verification_email(session: AsyncSession, email: str) -> str: - normalized_email = _normalize_email(email) - user = await _get_user_by_email(session, normalized_email) + require_delivery_configuration() - if user is None or user.primary_email_verified: - return "If the email is registered, a verification link has been sent" + try: + user = await session.scalar( + select(User) + .where(User.primary_email == _normalize_email(email)) + .with_for_update() + .execution_options(populate_existing=True), + ) - if not email_service.is_email_delivery_configured(): - return "Email verification is not configured on this server" + if user is not None: + await queue_verification(session, user) - token = await _issue_email_action_token( - session, - user.user_id, - EMAIL_VERIFICATION_ACTION, - get_settings().email_verification_token_expire_minutes, - ) + await session.commit() + except SQLAlchemyError as error: + await session.rollback() + raise VerificationUnavailableError() from error - email_service.send_email( - normalized_email, - "Verify your Papyrus email address", - _verification_email_body(token), - ) - - await session.commit() return "If the email is registered, a verification link has been sent" async def verify_email_token(session: AsyncSession, token: str) -> str: - email_token = await _consume_email_action_token(session, token, EMAIL_VERIFICATION_ACTION) - user = await _get_active_user(session, email_token.user_id) - user.primary_email_verified = True + try: + email_token = await _consume_email_action_token(session, token, EMAIL_VERIFICATION_ACTION) + except ValidationError as error: + raise VerificationTokenInvalidError() from error + + user = await session.get(User, email_token.user_id) + + if user is None or not accepts_verification(user) or email_token.email != user.primary_email: + await session.rollback() + raise VerificationTokenInvalidError() + + record_verification(user, "papyrus_email") + await cancel_verification_deliveries(session, user) await session.commit() return "Email verified successfully" diff --git a/papyrus/services/auth/google.py b/papyrus/services/auth/google.py index 7fa6801..bfb75a9 100644 --- a/papyrus/services/auth/google.py +++ b/papyrus/services/auth/google.py @@ -10,11 +10,11 @@ import jwt from sqlalchemy import select +from sqlalchemy.exc import IntegrityError, SQLAlchemyError from sqlalchemy.ext.asyncio import AsyncSession -from sqlalchemy.orm import selectinload from papyrus.config import get_settings -from papyrus.core.exceptions import ConflictError, ForbiddenError, ValidationError +from papyrus.core.exceptions import ConflictError, ForbiddenError, ValidationError, VerificationUnavailableError from papyrus.core.security import create_state_token, decode_state_token from papyrus.models import User, UserIdentity from papyrus.services.auth._core import ( @@ -23,10 +23,16 @@ _default_display_name, _get_active_user, _get_user_by_email, + _lock_credential_owner, _normalize_email, _now, ) from papyrus.services.auth.types import GOOGLE_PROVIDER, GoogleIdentity +from papyrus.services.auth.verification import ( + cancel_verification_deliveries, + queue_verification, + record_verification, +) GOOGLE_AUTHORIZATION_URL = "https://accounts.google.com/o/oauth2/v2/auth" GOOGLE_TOKEN_URL = "https://oauth2.googleapis.com/token" @@ -100,6 +106,7 @@ def exchange_code_for_identity(self, code: str, callback_uri: str) -> GoogleIden algorithms=["RS256"], audience=settings.google_oauth_client_id, issuer=["accounts.google.com", "https://accounts.google.com"], + options={"require": ["sub", "exp", "iss", "aud"]}, ) email = claims.get("email") @@ -107,7 +114,8 @@ def exchange_code_for_identity(self, code: str, callback_uri: str) -> GoogleIden return GoogleIdentity( subject=str(claims["sub"]), email=_normalize_email(email) if isinstance(email, str) else None, - email_verified=bool(claims.get("email_verified", False)), + email_verified=claims.get("email_verified") is True, + hosted_domain=claims.get("hd") if isinstance(claims.get("hd"), str) else None, display_name=str(claims["name"]) if isinstance(claims.get("name"), str) else None, avatar_url=str(claims["picture"]) if isinstance(claims.get("picture"), str) else None, nonce=claims.get("nonce"), @@ -214,57 +222,89 @@ async def build_google_link_authorization_url(user_id: UUID, redirect_uri: str, ) +def _google_is_authoritative(identity: GoogleIdentity) -> bool: + if identity.email is None: + return False + + return identity.email.endswith("@gmail.com") or ( + identity.email_verified and bool(identity.hosted_domain and identity.hosted_domain.strip()) + ) + + async def _resolve_google_login_user(session: AsyncSession, identity: GoogleIdentity) -> tuple[User | None, str | None]: - identity_result = await session.execute( - select(UserIdentity) - .options(selectinload(UserIdentity.user)) - .where( + if identity.email is not None: + identity.email = _normalize_email(identity.email) + + existing_identity = await session.scalar( + select(UserIdentity).where( UserIdentity.provider == GOOGLE_PROVIDER, UserIdentity.provider_subject == identity.subject, - ) + ), ) - existing_identity = identity_result.scalar_one_or_none() + new_account = existing_identity is None if existing_identity is not None: - if existing_identity.user.deletion_due_at is not None: - return existing_identity.user, None + await _lock_credential_owner(session, existing_identity.user_id) + user = await session.get(User, existing_identity.user_id) - if existing_identity.user.disabled_at is not None: + if user is None or (user.disabled_at is not None and user.deletion_due_at is None): return None, "account_disabled" + if user.deletion_due_at is not None and user.deletion_due_at <= _now(): + return None, "account_disabled" + + if user.primary_email is None: + if identity.email is None: + return None, "email_required" + + if await _get_user_by_email(session, identity.email) is not None: + return None, "account_exists" + + user.primary_email = identity.email + existing_identity.last_used_at = _now() existing_identity.email_at_provider = identity.email existing_identity.email_verified_at_provider = _now() if identity.email_verified else None - return existing_identity.user, None - - if identity.email is not None: - existing_user = await _get_user_by_email(session, identity.email) + existing_identity.hosted_domain = identity.hosted_domain + else: + if identity.email is None: + return None, "email_required" - if existing_user is not None: + if await _get_user_by_email(session, identity.email) is not None: return None, "account_exists" - user = User( - display_name=_default_display_name(identity, identity.email), - avatar_url=identity.avatar_url, - primary_email=identity.email, - primary_email_verified=identity.email_verified, - last_login_at=_now(), - ) - - session.add(user) - await session.flush() + user = User( + display_name=_default_display_name(identity, identity.email), + avatar_url=identity.avatar_url, + primary_email=identity.email, + primary_email_verified=False, + ) - session.add( - UserIdentity( - user_id=user.user_id, - provider=GOOGLE_PROVIDER, - provider_subject=identity.subject, - email_at_provider=identity.email, - email_verified_at_provider=_now() if identity.email_verified else None, - last_used_at=_now(), + session.add(user) + await session.flush() + + session.add( + UserIdentity( + user_id=user.user_id, + provider=GOOGLE_PROVIDER, + provider_subject=identity.subject, + email_at_provider=identity.email, + email_verified_at_provider=_now() if identity.email_verified else None, + hosted_domain=identity.hosted_domain, + last_used_at=_now(), + ), ) - ) + + if identity.email == user.primary_email and _google_is_authoritative(identity) and not user.email_verified: + record_verification(user, "google", identity.subject) + await cancel_verification_deliveries(session, user) + + if not user.email_verified: + queued = await queue_verification(session, user) + + if new_account and not queued: + raise VerificationUnavailableError() await session.flush() return user, None @@ -328,20 +368,30 @@ async def handle_google_callback( return _build_redirect_uri(redirect_uri, {"code": exchange_code}) if mode == "login": - user, callback_error = await _resolve_google_login_user(session, identity) + try: + user, callback_error = await _resolve_google_login_user(session, identity) - if callback_error is not None or user is None: - await session.rollback() - return _build_redirect_uri(redirect_uri, {"error": callback_error or "oauth_failed"}) + if callback_error is not None or user is None: + await session.rollback() + return _build_redirect_uri(redirect_uri, {"error": callback_error or "oauth_failed"}) - exchange_code = await _create_exchange_code( - session, - purpose="login", - redirect_uri=redirect_uri, - user_id=user.user_id, - ) + purpose = "login" if user.email_verified else "login_pending_verification" + + exchange_code = await _create_exchange_code( + session, + purpose=purpose, + redirect_uri=redirect_uri, + user_id=user.user_id, + ) + + await session.commit() + except IntegrityError: + await session.rollback() + return _build_redirect_uri(redirect_uri, {"error": "account_exists"}) + except (VerificationUnavailableError, SQLAlchemyError): + await session.rollback() + return _build_redirect_uri(redirect_uri, {"error": "verification_unavailable"}) - await session.commit() return _build_redirect_uri(redirect_uri, {"code": exchange_code}) if mode == "link": @@ -358,6 +408,7 @@ async def handle_google_callback( provider=GOOGLE_PROVIDER, provider_subject=identity.subject, email_at_provider=identity.email, + hosted_domain=identity.hosted_domain, email_verified_at_provider=_now() if identity.email_verified else None, display_name=identity.display_name, avatar_url=identity.avatar_url, @@ -403,6 +454,7 @@ async def complete_google_link(session: AsyncSession, user_id: UUID, code: str) provider=GOOGLE_PROVIDER, provider_subject=exchange_code.provider_subject, email_at_provider=exchange_code.email_at_provider, + hosted_domain=exchange_code.hosted_domain, email_verified_at_provider=exchange_code.email_verified_at_provider, last_used_at=_now(), ) @@ -410,7 +462,7 @@ async def complete_google_link(session: AsyncSession, user_id: UUID, code: str) if user.primary_email is None and exchange_code.email_at_provider is not None: user.primary_email = exchange_code.email_at_provider - user.primary_email_verified = exchange_code.email_verified_at_provider is not None + user.primary_email_verified = False if user.avatar_url is None and exchange_code.avatar_url is not None: user.avatar_url = exchange_code.avatar_url diff --git a/papyrus/services/auth/sessions.py b/papyrus/services/auth/sessions.py index afbd1d7..3a7ce2d 100644 --- a/papyrus/services/auth/sessions.py +++ b/papyrus/services/auth/sessions.py @@ -6,11 +6,12 @@ from uuid import UUID from sqlalchemy import select +from sqlalchemy.exc import IntegrityError, SQLAlchemyError from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import selectinload from papyrus.config import get_settings -from papyrus.core.exceptions import ConflictError, ForbiddenError, UnauthorizedError +from papyrus.core.exceptions import ConflictError, ForbiddenError, UnauthorizedError, VerificationUnavailableError from papyrus.core.security import ( create_access_token, generate_opaque_token, @@ -19,57 +20,74 @@ verify_password, ) from papyrus.models import AuthSession, PasswordCredential, User -from papyrus.schemas.auth import LoginRequest, OAuthExchangeRequest, RefreshTokenRequest, RegisterRequest +from papyrus.schemas.auth import ( + LoginRequest, + OAuthExchangeRequest, + PendingVerificationResponse, + RefreshTokenRequest, + RegisterRequest, +) from papyrus.services.auth._core import ( _consume_exchange_code, _create_session_for_user, _expires_in_seconds, _get_active_user, _get_user_by_email, + _lock_credential_owner, _normalize_email, _now, _revoke_user_sessions, ) from papyrus.services.auth.types import AuthResult +from papyrus.services.auth.verification import ( + queue_verification, + require_delivery_configuration, + verification_required, +) async def register_user( session: AsyncSession, request: RegisterRequest, user_agent: str | None, -) -> AuthResult: +) -> PendingVerificationResponse: + require_delivery_configuration() normalized_email = _normalize_email(request.email) - if await _get_user_by_email(session, normalized_email) is not None: - raise ConflictError("An account with this email already exists") + try: + if await _get_user_by_email(session, normalized_email) is not None: + raise ConflictError("An account with this email already exists") - user = User( - display_name=request.display_name, - primary_email=normalized_email, - primary_email_verified=False, - ) + user = User( + display_name=request.display_name, + primary_email=normalized_email, + primary_email_verified=False, + ) - session.add(user) - await session.flush() + session.add(user) + await session.flush() - session.add( - PasswordCredential( - user_id=user.user_id, - password_hash=hash_password(request.password), + session.add( + PasswordCredential(user_id=user.user_id, password_hash=hash_password(request.password)), ) - ) - result = await _create_session_for_user( - session, - user, - client_type=request.client_type, - device_label=request.device_label, - user_agent=user_agent, - ) + if not await queue_verification(session, user): + await session.rollback() + raise VerificationUnavailableError() - await session.commit() - await session.refresh(user) - return result + await session.commit() + except IntegrityError as error: + await session.rollback() + + if await _get_user_by_email(session, normalized_email) is not None: + raise ConflictError("An account with this email already exists") from error + + raise VerificationUnavailableError() from error + except SQLAlchemyError as error: + await session.rollback() + raise VerificationUnavailableError() from error + + return PendingVerificationResponse(email=normalized_email) async def login_user( @@ -81,6 +99,8 @@ async def login_user( select(User) .options(selectinload(User.password_credential)) .where(User.primary_email == _normalize_email(request.email)) + .with_for_update() + .execution_options(populate_existing=True) ) user = result.scalar_one_or_none() @@ -96,7 +116,7 @@ async def login_user( await require_account_recovery(session, user.user_id) - if user.disabled_at is not None: + if user.disabled_at is not None or user.deletion_due_at is not None: raise ForbiddenError("User account is disabled") auth_result = await _create_session_for_user( @@ -116,10 +136,12 @@ async def refresh_tokens( session: AsyncSession, request: RefreshTokenRequest, ) -> AuthResult: + query = select(AuthSession).where(AuthSession.refresh_token_hash == hash_opaque_token(request.refresh_token)) + user_id = await session.scalar(query.with_only_columns(AuthSession.user_id)) + await _lock_credential_owner(session, user_id) + result = await session.execute( - select(AuthSession) - .options(selectinload(AuthSession.user)) - .where(AuthSession.refresh_token_hash == hash_opaque_token(request.refresh_token)) + query.options(selectinload(AuthSession.user)).with_for_update().execution_options(populate_existing=True), ) auth_session = result.scalar_one_or_none() @@ -129,9 +151,15 @@ async def refresh_tokens( user = auth_session.user - if user.disabled_at is not None: + if user.disabled_at is not None or user.deletion_due_at is not None: raise ForbiddenError("User account is disabled") + if not user.email_verified: + error = await verification_required(session, user) + await _revoke_user_sessions(session, user.user_id) + await session.commit() + raise error + new_refresh_token = generate_opaque_token() auth_session.refresh_token_hash = hash_opaque_token(new_refresh_token) auth_session.expires_at = _now() + timedelta(days=get_settings().refresh_token_expire_days) @@ -173,7 +201,7 @@ async def exchange_login_code( request: OAuthExchangeRequest, user_agent: str | None, ) -> AuthResult: - exchange_code = await _consume_exchange_code(session, request.code, purpose="login") + exchange_code = await _consume_exchange_code(session, request.code, purpose=("login", "login_pending_verification")) if exchange_code.user_id is None: raise UnauthorizedError("Exchange code is missing a user") @@ -187,6 +215,11 @@ async def exchange_login_code( user = await _get_active_user(session, exchange_code.user_id) + if exchange_code.purpose == "login_pending_verification" or not user.email_verified: + error = await verification_required(session, user) + await session.commit() + raise error + auth_result = await _create_session_for_user( session, user, diff --git a/papyrus/services/auth/types.py b/papyrus/services/auth/types.py index 37594af..94b1b1b 100644 --- a/papyrus/services/auth/types.py +++ b/papyrus/services/auth/types.py @@ -31,3 +31,4 @@ class GoogleIdentity: display_name: str | None avatar_url: str | None nonce: str | None = None + hosted_domain: str | None = None diff --git a/papyrus/services/auth/verification.py b/papyrus/services/auth/verification.py new file mode 100644 index 0000000..eea6690 --- /dev/null +++ b/papyrus/services/auth/verification.py @@ -0,0 +1,226 @@ +"""Email-bound verification evidence and a dedicated transactional delivery outbox.""" + +import asyncio +import logging +import math +from datetime import timedelta +from urllib.parse import urlsplit + +from sqlalchemy import delete, select, update +from sqlalchemy.dialects.postgresql import insert +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker + +from papyrus.config import get_settings +from papyrus.core.exceptions import VerificationRequiredError, VerificationUnavailableError +from papyrus.core.security import generate_opaque_token, hash_opaque_token +from papyrus.models import EmailActionToken, User, VerificationDelivery, VerificationRecipient +from papyrus.services import email +from papyrus.services.auth._core import _build_app_url, _now +from papyrus.services.auth.types import EMAIL_VERIFICATION_ACTION + +COOLDOWN_SECONDS = 60 +TOKEN_LIFETIME = timedelta(hours=24) +logger = logging.getLogger(__name__) + + +def require_delivery_configuration() -> None: + base = get_settings().app_public_base_url + parsed = urlsplit(base or "") + + if not email.is_email_delivery_configured() or parsed.scheme not in {"http", "https"} or not parsed.netloc: + raise VerificationUnavailableError() + + +def accepts_verification(user: User) -> bool: + if user.primary_email is None: + return False + + if user.deletion_due_at is not None: + return user.deletion_due_at > _now() + + return user.disabled_at is None + + +def record_verification(user: User, source: str, subject: str | None = None) -> None: + user.primary_email_verified = True + user.verified_email = user.primary_email + user.verification_source = source + user.verification_subject = subject + user.verified_at = _now() + + +async def require_verified(session: AsyncSession, user: User) -> None: + if not user.email_verified: + raise await verification_required(session, user) + + +async def verification_required(session: AsyncSession, user: User) -> VerificationRequiredError: + recipient = await session.get(VerificationRecipient, user.primary_email) if user.primary_email else None + remaining = 0 + + if recipient is not None: + remaining = max(0, math.ceil((recipient.last_queued_at + timedelta(seconds=60) - _now()).total_seconds())) + + return VerificationRequiredError(user.primary_email, remaining) + + +async def queue_verification(session: AsyncSession, user: User) -> bool: + """The caller holds the owner lock and commits account, token and delivery together.""" + require_delivery_configuration() + + if user.email_verified or not accepts_verification(user): + return False + + now = _now() + + claimed = await session.scalar( + insert(VerificationRecipient) + .values(email=user.primary_email, last_queued_at=now) + .on_conflict_do_update( + index_elements=[VerificationRecipient.email], + set_={"last_queued_at": now}, + where=VerificationRecipient.last_queued_at <= now - timedelta(seconds=COOLDOWN_SECONDS), + ) + .returning(VerificationRecipient.email) + ) + + if claimed is None: + return False + + token = generate_opaque_token() + expires_at = now + TOKEN_LIFETIME + + action = EmailActionToken( + user_id=user.user_id, + email=user.primary_email, + action_type=EMAIL_VERIFICATION_ACTION, + token_hash=hash_opaque_token(token), + expires_at=expires_at, + ) + + session.add(action) + await session.flush() + + session.add( + VerificationDelivery( + token_id=action.token_id, + user_id=user.user_id, + recipient=user.primary_email, + token=token, + next_attempt_at=now, + expires_at=expires_at, + ), + ) + + await session.flush() + return True + + +async def cancel_verification_deliveries(session: AsyncSession, user: User) -> None: + await session.execute( + delete(VerificationDelivery).where( + VerificationDelivery.user_id == user.user_id, + VerificationDelivery.recipient == user.primary_email, + ), + ) + + await session.execute( + update(EmailActionToken) + .where( + EmailActionToken.user_id == user.user_id, + EmailActionToken.email == user.primary_email, + EmailActionToken.action_type == EMAIL_VERIFICATION_ACTION, + EmailActionToken.used_at.is_(None), + ) + .values(used_at=_now()), + ) + + +async def deliver_pending_verifications(session: AsyncSession) -> int: + """Lease before SMTP. Duplicate delivery keeps the original token and expiry. + + A send already accepted by SMTP cannot be recalled after concurrent verification + or deletion. Its token is invalidated by those transitions and cannot grant access. + Neither SMTP exceptions nor SQL parameters containing secrets are logged. + """ + valid_token = select(EmailActionToken.token_id).where( + EmailActionToken.used_at.is_(None), + EmailActionToken.expires_at > _now(), + EmailActionToken.email == User.primary_email, + EmailActionToken.user_id == User.user_id, + ((User.disabled_at.is_(None) & User.deletion_due_at.is_(None)) | (User.deletion_due_at > _now())), + ) + + await session.execute( + delete(VerificationDelivery).where( + (VerificationDelivery.expires_at <= _now()) | VerificationDelivery.token_id.not_in(valid_token), + ), + ) + + await session.execute( + delete(VerificationRecipient).where( + VerificationRecipient.last_queued_at <= _now() - timedelta(seconds=COOLDOWN_SECONDS), + ), + ) + + await session.commit() + + try: + require_delivery_configuration() + except VerificationUnavailableError: + return 0 + + delivered = 0 + + for _ in range(100): + now = _now() + + job = await session.scalar( + select(VerificationDelivery) + .where(VerificationDelivery.next_attempt_at <= now, VerificationDelivery.expires_at > now) + .order_by(VerificationDelivery.next_attempt_at) + .with_for_update(skip_locked=True) + .limit(1), + ) + + if job is None: + await session.rollback() + break + + job.attempts += 1 + job.next_attempt_at = min(now + timedelta(minutes=5), job.expires_at) + await session.commit() + url = _build_app_url("/verify-email", {"token": job.token}) + body = f"Verify your Papyrus email address.\n\n{url}\n\nThis link expires 24 hours after it was requested." + + try: + await asyncio.to_thread(email.send_email, job.recipient, "Verify your Papyrus email address", body) + except Exception: + retry_at = min(_now() + timedelta(seconds=min(60 * 2 ** min(job.attempts - 1, 6), 3600)), job.expires_at) + + await session.execute( + update(VerificationDelivery) + .where(VerificationDelivery.token_id == job.token_id) + .values(next_attempt_at=retry_at), + ) + + await session.commit() + logger.warning("Verification delivery failed and will be retried") + continue + + await session.execute(delete(VerificationDelivery).where(VerificationDelivery.token_id == job.token_id)) + await session.commit() + delivered += 1 + + return delivered + + +async def run_delivery(session_maker: async_sessionmaker[AsyncSession]) -> None: + while True: + try: + async with session_maker() as session: + await deliver_pending_verifications(session) + except Exception: + logger.warning("Verification delivery cycle failed and will be retried") + + await asyncio.sleep(30) diff --git a/tests/api/routes/test_auth.py b/tests/api/routes/test_auth.py index e225539..41c48c3 100644 --- a/tests/api/routes/test_auth.py +++ b/tests/api/routes/test_auth.py @@ -17,9 +17,10 @@ from papyrus.core.exceptions import ServiceUnavailableError from papyrus.core.rate_limit import limiter from papyrus.core.security import generate_opaque_token, hash_opaque_token -from papyrus.models import AuthExchangeCode, EmailActionToken, User, UserIdentity +from papyrus.models import AuthExchangeCode, EmailActionToken, User, UserIdentity, VerificationDelivery from papyrus.services import auth as auth_service from papyrus.services import email as email_service +from papyrus.services.auth.verification import deliver_pending_verifications, record_verification @pytest.fixture @@ -79,7 +80,7 @@ def unconfigured_powersync(monkeypatch: pytest.MonkeyPatch) -> None: security_module._get_powersync_public_key.cache_clear() -@pytest.fixture +@pytest.fixture(autouse=True) def configured_email_delivery(monkeypatch: pytest.MonkeyPatch) -> list[tuple[str, str, str, str | None]]: settings = get_settings() monkeypatch.setattr(settings, "email_delivery_enabled", True) @@ -103,7 +104,7 @@ def unconfigured_email_delivery(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setattr(settings, "smtp_from_email", None) -async def test_register_user_returns_tokens(client: AsyncClient): +async def test_register_user_returns_pending_verification(client: AsyncClient): """Test user registration endpoint.""" response = await client.post( "/v1/auth/register", @@ -117,10 +118,13 @@ async def test_register_user_returns_tokens(client: AsyncClient): assert response.status_code == 201 data = response.json() - assert "access_token" in data - assert "refresh_token" in data - assert data["user"]["email"] == "test@example.com" - assert data["user"]["display_name"] == "Test User" + assert data == { + "status": "verification_required", + "email": "test@example.com", + "message": "Check your email to verify your account, then sign in.", + "resend_after": 60, + } + assert response.headers["cache-control"] == "no-store" async def test_register_duplicate_email_returns_conflict(client: AsyncClient): @@ -137,7 +141,15 @@ async def test_register_duplicate_email_returns_conflict(client: AsyncClient): assert second_response.status_code == 409 -async def test_login_user(client: AsyncClient): +async def verify_and_login(client, db_session, email="test@example.com"): + delivery = await db_session.scalar(select(VerificationDelivery).where(VerificationDelivery.recipient == email)) + assert (await client.post("/v1/auth/verify-email", json={"token": delivery.token})).status_code == 200 + response = await client.post("/v1/auth/login", json={"email": email, "password": "SecureP@ss123"}) + assert response.status_code == 200 + return response + + +async def test_login_user(client: AsyncClient, db_session: AsyncSession): """Test user login endpoint.""" register_response = await client.post( "/v1/auth/register", @@ -149,6 +161,7 @@ async def test_login_user(client: AsyncClient): ) assert register_response.status_code == 201 + register_response = await verify_and_login(client, db_session) response = await client.post( "/v1/auth/login", @@ -191,7 +204,7 @@ async def test_login_user_rejects_invalid_password(client: AsyncClient): assert response.status_code == 401 -async def test_refresh_token_rotates_and_invalidates_previous_token(client: AsyncClient): +async def test_refresh_token_rotates_and_invalidates_previous_token(client: AsyncClient, db_session: AsyncSession): """Test refresh token rotation endpoint.""" register_response = await client.post( "/v1/auth/register", @@ -203,6 +216,7 @@ async def test_refresh_token_rotates_and_invalidates_previous_token(client: Asyn ) assert register_response.status_code == 201 + register_response = await verify_and_login(client, db_session) refresh_token = register_response.json()["refresh_token"] refresh_response = await client.post("/v1/auth/refresh", json={"refresh_token": refresh_token}) assert refresh_response.status_code == 200 @@ -212,7 +226,7 @@ async def test_refresh_token_rotates_and_invalidates_previous_token(client: Asyn assert invalidated_response.status_code == 401 -async def test_logout_current_session_revokes_refresh_token(client: AsyncClient): +async def test_logout_current_session_revokes_refresh_token(client: AsyncClient, db_session: AsyncSession): """Test user logout revokes the current session refresh token.""" register_response = await client.post( "/v1/auth/register", @@ -224,6 +238,7 @@ async def test_logout_current_session_revokes_refresh_token(client: AsyncClient) ) assert register_response.status_code == 201 + register_response = await verify_and_login(client, db_session) auth_payload = register_response.json() logout_response = await client.post( @@ -267,11 +282,11 @@ async def test_login_is_rate_limited(client: AsyncClient): for _ in range(6) ] - assert [response.status_code for response in responses[:5]] == [200] * 5 + assert [response.status_code for response in responses[:5]] == [403] * 5 assert responses[5].status_code == 429 -async def test_logout_all_revokes_other_sessions(client: AsyncClient): +async def test_logout_all_revokes_other_sessions(client: AsyncClient, db_session: AsyncSession): """Test logout-all revokes all refresh-token backed sessions.""" register_response = await client.post( "/v1/auth/register", @@ -283,6 +298,7 @@ async def test_logout_all_revokes_other_sessions(client: AsyncClient): ) assert register_response.status_code == 201 + register_response = await verify_and_login(client, db_session) first_auth = register_response.json() login_response = await client.post( @@ -351,6 +367,7 @@ async def test_google_oauth_flow_creates_user_and_exchanges_code( subject="google-sub-1", email="google_user@example.com", email_verified=True, + hosted_domain="example.com", display_name="Google User", avatar_url="https://example.com/avatar.png", ), @@ -624,6 +641,7 @@ async def test_verify_email(client: AsyncClient, db_session: AsyncSession): EmailActionToken( user_id=user.user_id, action_type=auth_service.EMAIL_VERIFICATION_ACTION, + email=user.primary_email, token_hash=hash_opaque_token(plain_token), expires_at=datetime.now(UTC) + timedelta(hours=1), ) @@ -646,6 +664,7 @@ async def test_verify_email_rejects_expired_token(client: AsyncClient, db_sessio EmailActionToken( user_id=user.user_id, action_type=auth_service.EMAIL_VERIFICATION_ACTION, + email=user.primary_email, token_hash=hash_opaque_token(plain_token), expires_at=datetime.now(UTC) - timedelta(minutes=1), ) @@ -654,11 +673,12 @@ async def test_verify_email_rejects_expired_token(client: AsyncClient, db_sessio await db_session.commit() response = await client.post("/v1/auth/verify-email", json={"token": plain_token}) assert response.status_code == 400 - assert response.json()["error"]["code"] == "VALIDATION_ERROR" + assert response.json()["error"]["code"] == "VERIFICATION_TOKEN_INVALID" async def test_resend_verification_sends_email_when_configured( client: AsyncClient, + db_session: AsyncSession, configured_email_delivery: list[tuple[str, str, str, str | None]], ): """Test resend verification sends an email when SMTP is configured.""" @@ -680,33 +700,25 @@ async def test_resend_verification_sends_email_when_configured( assert response.status_code == 200 assert response.json()["message"] == "If the email is registered, a verification link has been sent" + assert configured_email_delivery == [] + assert await deliver_pending_verifications(db_session) == 1 assert len(configured_email_delivery) == 1 recipient, subject, body, html = configured_email_delivery[0] assert recipient == "test@example.com" assert subject == "Verify your Papyrus email address" - assert "Verification token:" in body + assert "/verify-email?token=" in body assert html is None async def test_forgot_password_returns_configuration_message( client: AsyncClient, unconfigured_email_delivery: None, + auth_user: dict[str, str], ): """Test forgot password endpoint without SMTP configuration.""" - register_response = await client.post( - "/v1/auth/register", - json={ - "email": "test@example.com", - "password": "SecureP@ss123", - "display_name": "Test User", - }, - ) - - assert register_response.status_code == 201 - response = await client.post( "/v1/auth/forgot-password", - json={"email": "test@example.com"}, + json={"email": "user@example.com"}, ) assert response.status_code == 200 @@ -826,6 +838,7 @@ async def test_forgot_password_send_failure_returns_service_unavailable( async def test_reset_password(client: AsyncClient, db_session: AsyncSession): """Test password reset endpoint.""" user = User(display_name="Resettable User", primary_email="reset@example.com") + record_verification(user, "papyrus_email") db_session.add(user) await db_session.flush() plain_token = generate_opaque_token() @@ -899,6 +912,7 @@ async def test_reset_password_revokes_existing_access_token( async def test_reset_password_rejects_expired_token(client: AsyncClient, db_session: AsyncSession): """Test password reset rejects expired tokens.""" user = User(display_name="Resettable User", primary_email="reset@example.com") + record_verification(user, "papyrus_email") db_session.add(user) await db_session.flush() plain_token = generate_opaque_token() diff --git a/tests/api/routes/test_avatars.py b/tests/api/routes/test_avatars.py index bcafe0e..c21020f 100644 --- a/tests/api/routes/test_avatars.py +++ b/tests/api/routes/test_avatars.py @@ -64,13 +64,25 @@ async def test_photo_requires_authentication_and_owner( url = saved.json()["avatar_url"] assert (await client.get(f"/v1{url}")).status_code == 401 assert (await upload(client, {})).status_code == 401 - other = await client.post( - "/v1/auth/register", - json={"email": "other@example.com", "password": "SecureP@ss123", "display_name": "Other"}, + from papyrus.core.security import create_access_token, generate_opaque_token, hash_opaque_token + from papyrus.models import AuthSession + from papyrus.services.auth.verification import record_verification + + other = User(display_name="Other", primary_email="other@example.com") + record_verification(other, "papyrus_email") + db_session.add(other) + await db_session.flush() + + session = AuthSession( + user_id=other.user_id, + refresh_token_hash=hash_opaque_token(generate_opaque_token()), + expires_at=datetime.now(UTC) + timedelta(days=1), ) - assert other.status_code == 201 - other_headers = {"Authorization": f"Bearer {other.json()['access_token']}"} + db_session.add(session) + await db_session.commit() + token = create_access_token({"sub": str(other.user_id), "sid": str(session.session_id)}) + other_headers = {"Authorization": f"Bearer {token}"} assert (await client.get(f"/v1{url}", headers=other_headers)).status_code == 404 forged = await client.patch("/v1/users/me", headers=other_headers, json={"avatar_url": url}) assert forged.status_code == 400 diff --git a/tests/api/routes/test_rate_limits.py b/tests/api/routes/test_rate_limits.py index 48b25ab..328ad3c 100644 --- a/tests/api/routes/test_rate_limits.py +++ b/tests/api/routes/test_rate_limits.py @@ -63,6 +63,9 @@ async def test_upload_limits_are_per_validated_user_across_sessions( monkeypatch.setattr(settings, "media_storage_root", str(tmp_path)) monkeypatch.setattr(settings, "cors_origins", ["https://app.papyrus.test"]) other_user = User(display_name="Other", primary_email="rate-other@example.com", primary_email_verified=True) + other_user.verified_email = other_user.primary_email + other_user.verification_source = "papyrus_email" + other_user.verified_at = datetime.now(UTC) db_session.add(other_user) await db_session.flush() owner_id = UUID(auth_user["user_id"]) diff --git a/tests/api/routes/test_verification.py b/tests/api/routes/test_verification.py new file mode 100644 index 0000000..3e6658f --- /dev/null +++ b/tests/api/routes/test_verification.py @@ -0,0 +1,184 @@ +"""Public verification contracts, existing access guards, and generic retry responses.""" + +from datetime import UTC, datetime +from uuid import UUID + +import pytest +from sqlalchemy import func, select + +from papyrus.config import get_settings +from papyrus.models import AuthSession, User, VerificationDelivery, VerificationRecipient + + +@pytest.fixture(autouse=True) +def mail_configuration(monkeypatch): + settings = get_settings() + monkeypatch.setattr(settings, "email_delivery_enabled", True) + monkeypatch.setattr(settings, "smtp_host", "smtp.example.test") + monkeypatch.setattr(settings, "smtp_from_email", "noreply@example.test") + monkeypatch.setattr(settings, "app_public_base_url", "https://app.example.test") + + +@pytest.mark.parametrize( + "path,method,body", + [ + ("/v1/users/me", "get", None), + ("/v1/auth/powersync-token", "post", None), + ("/v1/sync/powersync-upload", "post", {"batch": []}), + ("/v1/media/00000000-0000-4000-8000-000000000001", "get", None), + ], +) +async def test_legacy_jwt_cannot_access_application(client, auth_user, auth_headers, db_session, path, method, body): + user = await db_session.get(User, UUID(auth_user["user_id"])) + user.verified_email = "old@example.com" + await db_session.commit() + response = await client.request(method, path, headers=auth_headers, json=body) + assert response.status_code == 403, response.text + assert response.json()["error"]["code"] == "EMAIL_VERIFICATION_REQUIRED" + assert response.headers["cache-control"] == "no-store" + assert "access_token" not in response.text + + +async def test_logout_remains_available_for_unverified_session(client, auth_user, auth_headers, db_session): + user = await db_session.get(User, UUID(auth_user["user_id"])) + user.verified_email = None + await db_session.commit() + assert (await client.post("/v1/auth/logout", headers=auth_headers)).status_code == 204 + await db_session.refresh(user) + stored = await db_session.get(AuthSession, UUID(auth_user["session_id"])) + assert stored.revoked_at is not None + + +@pytest.mark.parametrize("configuration", ["smtp", "landing"]) +async def test_unavailable_registration_and_resend_are_retryable_and_generic( + client, db_session, monkeypatch, configuration +): + if configuration == "smtp": + monkeypatch.setattr(get_settings(), "email_delivery_enabled", False) + else: + monkeypatch.setattr(get_settings(), "app_public_base_url", None) + + registration = await client.post( + "/v1/auth/register", json={"email": "reader@example.com", "password": "SecureP@ss123", "display_name": "Reader"} + ) + resend = await client.post("/v1/auth/resend-verification", json={"email": "missing@example.com"}) + + for response in (registration, resend): + assert response.status_code == 503 + assert response.headers["retry-after"] == "60" + assert response.headers["cache-control"] == "no-store" + assert response.json() == { + "error": { + "code": "VERIFICATION_UNAVAILABLE", + "message": "Verification email is temporarily unavailable. Please try again later.", + "details": {}, + } + } + + assert await db_session.scalar(select(User)) is None + assert await db_session.scalar(select(VerificationDelivery)) is None + + +async def test_resend_unknown_verified_disabled_and_pending_match(client, db_session, auth_user): + for address, disabled in [("pending@example.com", False), ("disabled@example.com", True)]: + db_session.add( + User(display_name="Reader", primary_email=address, disabled_at=datetime.now(UTC) if disabled else None) + ) + + await db_session.commit() + bodies = [] + + for address in [ + "missing@example.com", + "user@example.com", + "disabled@example.com", + "pending@example.com", + "PENDING@example.com", + ]: + response = await client.post("/v1/auth/resend-verification", json={"email": address}) + assert response.status_code == 200 + bodies.append(response.json()) + + assert ( + bodies == [{"message": "If the email is registered, a verification link has been sent", "resend_after": 60}] * 5 + ) + assert await db_session.scalar(select(func.count()).select_from(VerificationDelivery)) == 1 + + +@pytest.mark.parametrize("token", ["", "not-a-token", "x" * 2000]) +async def test_invalid_verification_envelope(client, token): + response = await client.post("/v1/auth/verify-email", json={"token": token}) + assert response.status_code == 400 + assert response.json()["error"] == { + "code": "VERIFICATION_TOKEN_INVALID", + "message": "This verification link is invalid or expired. Request a new link or sign in if you already verified.", + "details": {}, + } + + +async def test_registration_and_resend_share_cooldown_without_resetting_it(client, db_session): + response = await client.post( + "/v1/auth/register", json={"email": "Reader@example.com", "password": "SecureP@ss123", "display_name": "Reader"} + ) + assert response.status_code == 201 + before = await db_session.scalar(select(VerificationRecipient.last_queued_at)) + await client.post("/v1/auth/resend-verification", json={"email": "reader@example.com"}) + assert await db_session.scalar(select(VerificationRecipient.last_queued_at)) == before + assert await db_session.scalar(select(func.count()).select_from(VerificationDelivery)) == 1 + + +async def test_rate_limit_caps_configuration_and_matches_retry_header(client, monkeypatch): + monkeypatch.setattr(get_settings(), "rate_limit_auth", 100) + + for _ in range(5): + response = await client.post("/v1/auth/login", json={"email": "missing@example.com", "password": "wrong"}) + assert response.status_code == 401 + + response = await client.post("/v1/auth/login", json={"email": "missing@example.com", "password": "wrong"}) + assert response.status_code == 429 + assert int(response.headers["retry-after"]) == response.json()["error"]["details"]["retry_after"] + assert response.headers["cache-control"] == "no-store" + + +async def test_registration_uses_configured_api_prefix(monkeypatch, test_session_maker): + from httpx import ASGITransport, AsyncClient + + from papyrus.core.database import get_db + from papyrus.main import create_app + + monkeypatch.setattr(get_settings(), "api_prefix", "/custom/api") + app = create_app() + + async def database(): + async with test_session_maker() as session: + yield session + + app.dependency_overrides[get_db] = database + + async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client: + response = await client.post( + "/custom/api/auth/register", + json={"email": "reader@example.com", "password": "SecureP@ss123", "display_name": "Reader"}, + ) + assert response.status_code == 201 + assert response.headers["cache-control"] == "no-store" + assert response.json()["status"] == "verification_required" + + +def test_access_logs_redact_oauth_secrets(): + import logging + + from papyrus.main import AuthQueryFilter + + record = logging.LogRecord( + "uvicorn.access", + logging.INFO, + "", + 0, + '%s - "%s %s HTTP/%s" %d', + ("client", "GET", "/custom/auth/oauth/google/callback?code=secret&state=secret", "1.1", 302), + None, + ) + assert AuthQueryFilter().filter(record) + assert "secret" not in record.getMessage() + assert "/custom/auth/oauth/google/callback" in record.getMessage() diff --git a/tests/conftest.py b/tests/conftest.py index 3d88338..8da6b5e 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -303,6 +303,9 @@ async def auth_user( display_name="Example User", primary_email="user@example.com", primary_email_verified=True, + verified_email="user@example.com", + verification_source="papyrus_email", + verified_at=datetime.now(UTC), last_login_at=datetime.now(UTC), ) diff --git a/tests/services/test_auth.py b/tests/services/test_auth.py index 5aee6e6..85f2e14 100644 --- a/tests/services/test_auth.py +++ b/tests/services/test_auth.py @@ -14,9 +14,18 @@ from papyrus.core import security as security_module from papyrus.core.exceptions import ConflictError, ForbiddenError, UnauthorizedError, ValidationError from papyrus.core.security import generate_opaque_token, hash_opaque_token, hash_password -from papyrus.models import AuthExchangeCode, AuthSession, EmailActionToken, PasswordCredential, User, UserIdentity +from papyrus.models import ( + AuthExchangeCode, + AuthSession, + EmailActionToken, + PasswordCredential, + User, + UserIdentity, + VerificationDelivery, +) from papyrus.schemas.auth import LoginRequest, OAuthExchangeRequest, RegisterRequest from papyrus.services import auth as auth_service +from papyrus.services.auth.verification import record_verification def test_google_login_allows_configured_app_public_base_url_redirect( @@ -53,6 +62,7 @@ async def _create_user_with_password( disabled_at=datetime.now(UTC) if disabled else None, ) + record_verification(user, "papyrus_email") session.add(user) await session.flush() session.add(PasswordCredential(user_id=user.user_id, password_hash=hash_password(password))) @@ -61,6 +71,24 @@ async def _create_user_with_password( return user +@pytest.fixture(autouse=True) +def verification_configuration(monkeypatch): + settings = get_settings() + monkeypatch.setattr(settings, "email_delivery_enabled", True) + monkeypatch.setattr(settings, "smtp_host", "smtp.example.test") + monkeypatch.setattr(settings, "smtp_from_email", "noreply@example.test") + monkeypatch.setattr(settings, "app_public_base_url", "https://app.example.test") + + +async def register_verified(session, request, user_agent): + await auth_service.register_user(session, request, user_agent) + delivery = await session.scalar(select(VerificationDelivery).where(VerificationDelivery.recipient == request.email)) + await auth_service.verify_email_token(session, delivery.token) + return await auth_service.login_user( + session, LoginRequest(email=request.email, password=request.password), user_agent + ) + + @pytest.fixture def configured_google(monkeypatch: pytest.MonkeyPatch) -> None: settings = get_settings() @@ -83,10 +111,10 @@ def configured_powersync(monkeypatch: pytest.MonkeyPatch) -> None: security_module._get_powersync_public_key.cache_clear() -async def test_register_user_creates_user_and_session( +async def test_register_user_creates_pending_account_and_delivery( test_session_maker: async_sessionmaker[AsyncSession], ): - """Test registration creates a user and backing auth session.""" + """Registration atomically creates pending data without a session.""" async with test_session_maker() as session: result = await auth_service.register_user( session, @@ -99,12 +127,10 @@ async def test_register_user_creates_user_and_session( "pytest", ) - session_result = await session.execute(select(AuthSession).where(AuthSession.user_id == result.user.user_id)) - auth_session = session_result.scalar_one() - assert result.user.primary_email == "register@example.com" - assert result.refresh_token - assert auth_session.client_type == "web" - assert auth_session.revoked_at is None + assert result.email == "register@example.com" + assert result.status == "verification_required" + assert await session.scalar(select(AuthSession)) is None + assert await session.scalar(select(VerificationDelivery)) is not None async def test_register_user_rejects_duplicate_email( @@ -154,7 +180,7 @@ async def test_logout_current_session_revokes_session( ): """Test logout_current_session revokes only the targeted session.""" async with test_session_maker() as session: - result = await auth_service.register_user( + result = await register_verified( session, RegisterRequest( email="logout@example.com", @@ -176,7 +202,7 @@ async def test_logout_all_sessions_revokes_every_session( ): """Test logout_all_sessions revokes all sessions for the user.""" async with test_session_maker() as session: - await auth_service.register_user( + await register_verified( session, RegisterRequest( email="all-sessions@example.com", @@ -203,7 +229,7 @@ async def test_reset_password_revokes_all_sessions( ): """Test password reset revokes active sessions.""" async with test_session_maker() as session: - register_result = await auth_service.register_user( + register_result = await register_verified( session, RegisterRequest( email="reset@example.com", @@ -247,6 +273,7 @@ async def test_google_login_reuses_existing_identity( primary_email_verified=True, ) + record_verification(user, "papyrus_email") session.add(user) await session.flush() @@ -418,12 +445,14 @@ async def test_verify_email_token_rejects_expired_and_reused_tokens( EmailActionToken( user_id=user.user_id, action_type=auth_service.EMAIL_VERIFICATION_ACTION, + email=user.primary_email, token_hash=hash_opaque_token(expired_token), expires_at=datetime.now(UTC) - timedelta(minutes=1), ), EmailActionToken( user_id=user.user_id, action_type=auth_service.EMAIL_VERIFICATION_ACTION, + email=user.primary_email, token_hash=hash_opaque_token(valid_token), expires_at=datetime.now(UTC) + timedelta(minutes=30), ), diff --git a/tests/services/test_auth_recovery_concurrency.py b/tests/services/test_auth_recovery_concurrency.py index 7f21be3..f104999 100644 --- a/tests/services/test_auth_recovery_concurrency.py +++ b/tests/services/test_auth_recovery_concurrency.py @@ -8,14 +8,15 @@ from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker from papyrus.core.exceptions import UnauthorizedError, ValidationError -from papyrus.models import AuthSession, User +from papyrus.core.security import hash_opaque_token +from papyrus.models import AuthSession, EmailActionToken, User from papyrus.schemas.account_deletion import AccountPasswordRequest from papyrus.schemas.auth import OAuthExchangeRequest from papyrus.services import account_deletion, auth from papyrus.services.auth import _core -@pytest.mark.parametrize("flow", ["google", "password_reset"]) +@pytest.mark.parametrize("flow", ["google", "password_reset", "verification"]) async def test_cancellation_revokes_waiting_authentication_without_deadlock( test_session_maker: async_sessionmaker[AsyncSession], auth_user: dict[str, str], @@ -33,6 +34,12 @@ async def test_cancellation_revokes_waiting_authentication_without_deadlock( if flow == "google": token = await _core._create_exchange_code(setup, purpose="login", redirect_uri="", user_id=user_id) + elif flow == "verification": + token = await _core._issue_email_action_token(setup, user_id, auth.EMAIL_VERIFICATION_ACTION, 5) + stored = await setup.scalar( + select(EmailActionToken).where(EmailActionToken.token_hash == hash_opaque_token(token)) + ) + stored.email = "user@example.com" else: token = await _core._issue_email_action_token(setup, user_id, auth.PASSWORD_RESET_ACTION, 5) @@ -64,6 +71,9 @@ async def authenticate() -> None: if flow == "google": with pytest.raises(UnauthorizedError): await auth.exchange_login_code(authentication, OAuthExchangeRequest(code=token), None) + elif flow == "verification": + with pytest.raises(ValidationError): + await auth.verify_email_token(authentication, token) else: with pytest.raises(ValidationError): await auth.reset_password(authentication, token, "RacingP@ssword123") diff --git a/tests/services/test_verification.py b/tests/services/test_verification.py new file mode 100644 index 0000000..b84457d --- /dev/null +++ b/tests/services/test_verification.py @@ -0,0 +1,439 @@ +"""Verification evidence, delivery, and concurrency regressions against PostgreSQL.""" + +import asyncio +from datetime import UTC, datetime, timedelta +from urllib.parse import parse_qs, urlsplit + +import pytest +from sqlalchemy import func, select, update +from sqlalchemy.exc import SQLAlchemyError + +from papyrus.config import get_settings +from papyrus.core.exceptions import ( + AppError, + VerificationRequiredError, + VerificationTokenInvalidError, + VerificationUnavailableError, +) +from papyrus.core.security import create_state_token, hash_opaque_token +from papyrus.models import ( + AuthExchangeCode, + AuthSession, + EmailActionToken, + User, + UserIdentity, + VerificationDelivery, + VerificationRecipient, +) +from papyrus.schemas.auth import LoginRequest, OAuthExchangeRequest, RefreshTokenRequest, RegisterRequest +from papyrus.services import account_deletion, auth, email +from papyrus.services.auth import verification + + +@pytest.fixture(autouse=True) +def mail_configuration(monkeypatch): + settings = get_settings() + monkeypatch.setattr(settings, "email_delivery_enabled", True) + monkeypatch.setattr(settings, "smtp_host", "smtp.example.test") + monkeypatch.setattr(settings, "smtp_from_email", "noreply@example.test") + monkeypatch.setattr(settings, "app_public_base_url", "https://app.example.test/reader") + + +async def register(session, address="reader@example.com"): + await auth.register_user( + session, RegisterRequest(email=address, password="SecureP@ss123", display_name="Reader"), None + ) + return await session.scalar(select(User).where(User.primary_email == address)) + + +async def test_pending_denial_and_explicit_verification(test_session_maker): + async with test_session_maker() as session: + user = await register(session) + assert await session.scalar(select(AuthSession)) is None + + with pytest.raises(VerificationRequiredError) as denied: + await auth.login_user(session, LoginRequest(email=user.primary_email, password="SecureP@ss123"), None) + + assert denied.value.details == {"email": user.primary_email, "resend_after": 60} + job = await session.scalar(select(VerificationDelivery)) + token = await session.get(EmailActionToken, job.token_id) + assert token.email == user.primary_email + assert token.token_hash == hash_opaque_token(job.token) + assert 86395 <= (token.expires_at - datetime.now(UTC)).total_seconds() <= 86400 + await auth.verify_email_token(session, job.token) + assert await session.scalar(select(AuthSession)) is None + result = await auth.login_user(session, LoginRequest(email=user.primary_email, password="SecureP@ss123"), None) + assert result.user.email_verified + assert result.user.verification_source == "papyrus_email" + assert await session.scalar(select(VerificationDelivery)) is None + + +@pytest.mark.parametrize("state", ["expired", "used", "wrong_address", "disabled", "past_deletion", "unbound"]) +async def test_invalid_verification_is_uniform(test_session_maker, state): + async with test_session_maker() as session: + user = await register(session) + job = await session.scalar(select(VerificationDelivery)) + token = await session.get(EmailActionToken, job.token_id) + plain = job.token + user_id = user.user_id + + if state == "expired": + token.expires_at = datetime.now(UTC) - timedelta(seconds=1) + elif state == "used": + token.used_at = datetime.now(UTC) + elif state == "wrong_address": + user.primary_email = "changed@example.com" + elif state == "disabled": + user.disabled_at = datetime.now(UTC) + elif state == "past_deletion": + user.deletion_due_at = datetime.now(UTC) - timedelta(seconds=1) + else: + token.email = None + + await session.commit() + + with pytest.raises(VerificationTokenInvalidError): + await auth.verify_email_token(session, plain) + + async with test_session_maker() as session: + assert await session.scalar(select(AuthSession)) is None + assert not (await session.get(User, user_id)).email_verified + + +async def test_resend_keeps_links_and_consumes_all_on_success(test_session_maker): + async with test_session_maker() as session: + user = await register(session) + first = await session.scalar(select(VerificationDelivery)) + first_token = first.token + recipient = await session.get(VerificationRecipient, user.primary_email) + recipient.last_queued_at -= timedelta(seconds=61) + await session.commit() + await auth.resend_verification_email(session, user.primary_email) + assert await session.scalar(select(func.count()).select_from(VerificationDelivery)) == 2 + await auth.verify_email_token(session, first_token) + assert await session.scalar(select(VerificationDelivery)) is None + assert await session.scalar(select(EmailActionToken).where(EmailActionToken.used_at.is_(None))) is None + + with pytest.raises(VerificationTokenInvalidError): + await auth.verify_email_token(session, first_token) + + +async def test_cooldown_and_consumption_serialize_across_connections(test_session_maker): + async with test_session_maker() as session: + user = await register(session) + await session.execute( + update(VerificationRecipient).values(last_queued_at=datetime.now(UTC) - timedelta(seconds=61)) + ) + await session.commit() + + async def resend(): + async with test_session_maker() as session: + await auth.resend_verification_email(session, user.primary_email.upper()) + + await asyncio.gather(resend(), resend(), resend()) + + async with test_session_maker() as session: + assert await session.scalar(select(func.count()).select_from(VerificationDelivery)) == 2 + tokens = list((await session.scalars(select(VerificationDelivery.token))).all()) + + async def verify(token): + async with test_session_maker() as session: + try: + await auth.verify_email_token(session, token) + return True + except VerificationTokenInvalidError: + return False + + assert sorted(await asyncio.gather(*(verify(token) for token in tokens))) == [False, True] + + +async def test_delivery_failure_retries_same_secret_without_extending_expiry(test_session_maker, monkeypatch, caplog): + sent = [] + + def fail(*args, **kwargs): + raise RuntimeError("SMTP secret must not be logged") + + monkeypatch.setattr(email, "send_email", fail) + + async with test_session_maker() as session: + await register(session) + job = await session.scalar(select(VerificationDelivery)) + original_token, original_expiry = job.token, job.expires_at + assert await verification.deliver_pending_verifications(session) == 0 + await session.refresh(job) + assert job.attempts == 1 + assert job.next_attempt_at > datetime.now(UTC) + assert job.token == original_token and job.expires_at == original_expiry + assert original_token not in caplog.text + assert "SMTP secret" not in caplog.text + job.next_attempt_at = datetime.now(UTC) - timedelta(seconds=1) + await session.commit() + monkeypatch.setattr(email, "send_email", lambda *args: sent.append(args)) + assert await verification.deliver_pending_verifications(session) == 1 + assert f"https://app.example.test/reader/verify-email?token={original_token}" in sent[0][2] + assert await session.scalar(select(VerificationDelivery)) is None + token = await session.scalar(select(EmailActionToken)) + assert token.expires_at == original_expiry + + +async def test_expired_delivery_removed_without_smtp(test_session_maker, monkeypatch): + monkeypatch.setattr(email, "send_email", lambda *args: pytest.fail("expired mail sent")) + + async with test_session_maker() as session: + await register(session) + await session.execute(update(VerificationDelivery).values(expires_at=datetime.now(UTC) - timedelta(seconds=1))) + await session.commit() + assert await verification.deliver_pending_verifications(session) == 0 + assert await session.scalar(select(VerificationDelivery)) is None + + +async def test_queue_failure_rolls_back_registration(test_session_maker, monkeypatch): + from papyrus.services.auth import sessions + + async def fail(session, user): + await verification.queue_verification(session, user) + raise SQLAlchemyError("queue transaction failed") + + monkeypatch.setattr(sessions, "queue_verification", fail) + + async with test_session_maker() as session: + with pytest.raises(VerificationUnavailableError): + await register(session) + + assert await session.scalar(select(User)) is None + assert await session.scalar(select(EmailActionToken)) is None + assert await session.scalar(select(VerificationDelivery)) is None + assert await session.scalar(select(VerificationRecipient)) is None + + +@pytest.mark.parametrize( + "address,verified,hd,eligible", + [ + ("reader@gmail.com", False, None, True), + ("reader@workspace.example", True, "workspace.example", True), + ("reader@workspace.example", False, "workspace.example", False), + ("reader@workspace.example", True, None, False), + ("reader@other.example", True, "", False), + ], +) +async def test_google_authority_and_pending_handoff(test_session_maker, monkeypatch, address, verified, hd, eligible): + identity = auth.GoogleIdentity("subject", address, verified, "Reader", None, hosted_domain=hd) + monkeypatch.setattr(auth.google_oauth_client, "exchange_code_for_identity", lambda *args: identity) + + async with test_session_maker() as session: + redirect = await auth.handle_google_callback( + session, + callback_uri="https://api.example.test/v1/auth/oauth/google/callback", + code="provider", + state_token=create_state_token({"mode": "login", "redirect_uri": "papyrus://auth/callback"}), + error=None, + ) + + code = parse_qs(urlsplit(redirect).query)["code"][0] + assert await session.scalar(select(AuthSession)) is None + stored = await session.scalar(select(UserIdentity)) + assert stored.hosted_domain == hd + + if eligible: + result = await auth.exchange_login_code(session, OAuthExchangeRequest(code=code), None) + assert result.user.email_verified + assert result.user.verification_subject == "subject" + else: + job = await session.scalar(select(VerificationDelivery)) + await auth.verify_email_token(session, job.token) + + with pytest.raises(VerificationRequiredError): + await auth.exchange_login_code(session, OAuthExchangeRequest(code=code), None) + + assert await session.scalar(select(AuthSession)) is None + + with pytest.raises(AppError) as reused: + await auth.exchange_login_code(session, OAuthExchangeRequest(code=code), None) + + assert reused.value.status_code == 401 + + +@pytest.mark.parametrize( + "address,expected", + [(None, "email_required"), ("reader@other.example", "verification_unavailable"), ("reader@gmail.com", None)], +) +async def test_google_missing_email_and_delivery_outage(test_session_maker, monkeypatch, address, expected): + monkeypatch.setattr(get_settings(), "email_delivery_enabled", False) + identity = auth.GoogleIdentity("subject", address, True, "Reader", None) + monkeypatch.setattr(auth.google_oauth_client, "exchange_code_for_identity", lambda *args: identity) + + async with test_session_maker() as session: + redirect = await auth.handle_google_callback( + session, + callback_uri="https://api.example.test/v1/auth/oauth/google/callback", + code="provider", + state_token=create_state_token({"mode": "login", "redirect_uri": "papyrus://auth/callback"}), + error=None, + ) + + query = parse_qs(urlsplit(redirect).query) + + if expected: + assert query == {"error": [expected]} + assert await session.scalar(select(User)) is None + assert await session.scalar(select(AuthExchangeCode)) is None + else: + assert "code" in query + + +async def test_legacy_refresh_revocation_commits_and_rotation_serializes(test_session_maker, auth_user): + async with test_session_maker() as session: + user = await session.scalar(select(User)) + user.verified_email = None + await session.commit() + + with pytest.raises(VerificationRequiredError): + await auth.refresh_tokens(session, RefreshTokenRequest(refresh_token=auth_user["refresh_token"])) + + async with test_session_maker() as session: + stored = await session.scalar(select(AuthSession)) + assert stored.revoked_at is not None + + with pytest.raises(AppError) as retry: + await auth.refresh_tokens(session, RefreshTokenRequest(refresh_token=auth_user["refresh_token"])) + + assert retry.value.status_code == 401 + + +async def test_concurrent_refresh_only_one_succeeds(test_session_maker, auth_user): + async def refresh(): + async with test_session_maker() as session: + try: + await auth.refresh_tokens(session, RefreshTokenRequest(refresh_token=auth_user["refresh_token"])) + return True + except AppError: + return False + + assert sorted(await asyncio.gather(refresh(), refresh())) == [False, True] + + +async def test_recovery_requires_evidence_without_consuming_authorization(test_session_maker, auth_user): + from papyrus.schemas.account_deletion import AccountPasswordRequest + + async with test_session_maker() as session: + proof = await account_deletion.authenticate_password( + session, AccountPasswordRequest(email="user@example.com", password=auth_user["password"]) + ) + await account_deletion.schedule_deletion(session, proof.authorization_token) + user = await session.scalar(select(User)) + user.verified_email = None + await session.commit() + proof = await account_deletion.authenticate_password( + session, AccountPasswordRequest(email="user@example.com", password=auth_user["password"]) + ) + + with pytest.raises(VerificationRequiredError): + await account_deletion.cancel_deletion(session, proof.authorization_token) + + authorization = await session.scalar( + select(AuthExchangeCode).where(AuthExchangeCode.code_hash == hash_opaque_token(proof.authorization_token)) + ) + assert authorization.used_at is None + assert user.deletion_due_at is not None + await auth.resend_verification_email(session, user.primary_email) + job = await session.scalar(select(VerificationDelivery)) + await auth.verify_email_token(session, job.token) + assert user.deletion_due_at is not None and user.disabled_at is not None + result = await account_deletion.cancel_deletion(session, proof.authorization_token) + assert result.user.email_verified + + +@pytest.mark.parametrize( + "primary,provider_email,expected", + [ + ("original@example.com", "changed@gmail.com", "pending"), + (None, None, "email_required"), + (None, "occupied@gmail.com", "account_exists"), + (None, "available@gmail.com", "verified"), + ], +) +async def test_linked_google_preserves_owner_and_primary_email( + test_session_maker, monkeypatch, primary, provider_email, expected +): + identity = auth.GoogleIdentity("linked", provider_email, True, "Provider", None) + monkeypatch.setattr(auth.google_oauth_client, "exchange_code_for_identity", lambda *args: identity) + + async with test_session_maker() as session: + user = User(display_name="Original", primary_email=primary) + session.add(user) + session.add(User(display_name="Other", primary_email="occupied@gmail.com")) + await session.flush() + user_id = user.user_id + session.add(UserIdentity(user_id=user_id, provider="google", provider_subject="linked")) + await session.commit() + + redirect = await auth.handle_google_callback( + session, + callback_uri="https://api.example.test/v1/auth/oauth/google/callback", + code="provider", + state_token=create_state_token({"mode": "login", "redirect_uri": "papyrus://auth/callback"}), + error=None, + ) + + query = parse_qs(urlsplit(redirect).query) + stored = await session.get(User, user_id) + assert await session.scalar(select(func.count()).select_from(User)) == 2 + assert stored.display_name == "Original" + + if expected in {"account_exists", "email_required"}: + assert query == {"error": [expected]} + assert stored.primary_email == primary + assert await session.scalar(select(AuthExchangeCode)) is None + elif expected == "pending": + job = await session.scalar(select(VerificationDelivery)) + assert job.recipient == primary + assert stored.primary_email == primary + assert not stored.email_verified + else: + assert stored.primary_email == provider_email + assert stored.email_verified + + +async def test_delivery_lease_prevents_two_workers_sending_the_same_job(test_session_maker, monkeypatch): + sent = [] + monkeypatch.setattr(email, "send_email", lambda *args: sent.append(args)) + + async with test_session_maker() as session: + await register(session) + + async def deliver(): + async with test_session_maker() as session: + return await verification.deliver_pending_verifications(session) + + assert sum(await asyncio.gather(deliver(), deliver())) == 1 + assert len(sent) == 1 + + +async def test_duplicate_registration_race_preserves_original_account(test_session_maker): + async def create(): + async with test_session_maker() as session: + try: + await register(session) + return 201 + except AppError as error: + return error.status_code + + assert sorted(await asyncio.gather(create(), create())) == [201, 409] + + async with test_session_maker() as session: + assert await session.scalar(select(func.count()).select_from(User)) == 1 + assert await session.scalar(select(func.count()).select_from(VerificationDelivery)) == 1 + assert await session.scalar(select(AuthSession)) is None + + +async def test_new_account_cannot_commit_without_its_delivery_during_recipient_cooldown(test_session_maker): + async with test_session_maker() as session: + session.add(VerificationRecipient(email="reader@example.com", last_queued_at=datetime.now(UTC))) + await session.commit() + + with pytest.raises(VerificationUnavailableError): + await register(session) + + assert await session.scalar(select(User)) is None + assert await session.scalar(select(EmailActionToken)) is None + assert await session.scalar(select(VerificationDelivery)) is None diff --git a/tests/test_models.py b/tests/test_models.py index 12ebe69..195247b 100644 --- a/tests/test_models.py +++ b/tests/test_models.py @@ -93,6 +93,10 @@ def test_auth_models_are_registered_with_metadata() -> None: "avatar_url", "primary_email", "primary_email_verified", + "verified_email", + "verification_source", + "verification_subject", + "verified_at", "created_at", "last_login_at", "disabled_at", diff --git a/tests/test_openapi_export.py b/tests/test_openapi_export.py index 9556dcc..477d444 100644 --- a/tests/test_openapi_export.py +++ b/tests/test_openapi_export.py @@ -27,5 +27,8 @@ def test_export_ignores_environment_and_dev_routes(tmp_path: Path) -> None: schema = json.loads(baseline) assert all(not path.startswith("/__dev") for path in schema["paths"]) assert {"/v1/books", "/v1/auth/login", "/v1/opds/relay", "/v1/sync/powersync-upload"} <= schema["paths"].keys() + registration = schema["paths"]["/v1/auth/register"]["post"]["responses"]["201"] + assert registration["content"]["application/json"]["schema"]["$ref"].endswith("/PendingVerificationResponse") + assert "access_token" not in schema["components"]["schemas"]["PendingVerificationResponse"]["properties"] assert schema["info"]["contact"]["url"] == "https://github.com/PapyrusReader/papyrus" subprocess.run([*command, "--check"], check=True) diff --git a/tests/test_verification_migration.py b/tests/test_verification_migration.py new file mode 100644 index 0000000..489500f --- /dev/null +++ b/tests/test_verification_migration.py @@ -0,0 +1,126 @@ +"""Evidence backfill and session preservation on the isolated database.""" + +import importlib.util +from pathlib import Path +from uuid import uuid4 + +import pytest +from alembic.autogenerate import compare_metadata +from alembic.migration import MigrationContext +from alembic.operations import Operations +from sqlalchemy import text + +from papyrus.models import Base + +pytestmark = pytest.mark.migration + + +async def test_verification_migration_preserves_trustworthy_evidence_and_data(db_session, caplog): + path = Path(__file__).parents[1] / "alembic/versions/d1e2f3456789_email_verification_evidence.py" + spec = importlib.util.spec_from_file_location("verification_revision", path) + revision = importlib.util.module_from_spec(spec) + spec.loader.exec_module(revision) + + await db_session.execute(text("DROP TABLE verification_deliveries, verification_recipients")) + + for column in ("verified_email", "verification_source", "verification_subject", "verified_at"): + await db_session.execute(text(f"ALTER TABLE users DROP COLUMN {column}")) + + for table in ("user_identities", "auth_exchange_codes"): + await db_session.execute(text(f"ALTER TABLE {table} DROP COLUMN hosted_domain")) + + await db_session.execute(text("ALTER TABLE email_action_tokens DROP COLUMN email")) + identifiers = [uuid4() for _ in range(5)] + + for index, user_id in enumerate(identifiers): + await db_session.execute( + text( + "INSERT INTO users (user_id, display_name, primary_email, primary_email_verified) VALUES (:id, 'Preserved', :email, :verified)" + ), + { + "id": user_id, + "email": f"reader{index}@gmail.com" if index == 4 else f"reader{index}@example.com", + "verified": index != 3, + }, + ) + await db_session.execute( + text("INSERT INTO password_credentials (user_id, password_hash) VALUES (:id, 'unchanged-hash')"), + {"id": user_id}, + ) + await db_session.execute( + text( + "INSERT INTO auth_sessions (session_id, user_id, refresh_token_hash, expires_at) VALUES (:sid, :id, :hash, now() + interval '1 day')" + ), + {"sid": uuid4(), "id": user_id, "hash": str(index)}, + ) + + if index in (1, 2, 4): + await db_session.execute( + text( + "INSERT INTO user_identities (identity_id, user_id, provider, provider_subject, email_at_provider, email_verified_at_provider) VALUES (:identity, :id, 'google', :sub, :email, now())" + ), + {"identity": uuid4(), "id": user_id, "sub": str(index), "email": f"reader{index}@gmail.com"}, + ) + + await db_session.execute( + text( + "INSERT INTO books (book_id, owner_user_id, title, current_cfi) VALUES (:book, :owner, 'Durable library', 'epub:0:123')" + ), + {"book": uuid4(), "owner": identifiers[1]}, + ) + + def upgrade(session): + with Operations.context(MigrationContext.configure(session.connection())): + revision.upgrade() + + reader_role = "test_verification_" + uuid4().hex + await db_session.execute(text(f'CREATE ROLE "{reader_role}"')) + await db_session.execute( + text(f'ALTER DEFAULT PRIVILEGES IN SCHEMA public GRANT SELECT ON TABLES TO "{reader_role}"') + ) + await db_session.execute(text("ALTER DEFAULT PRIVILEGES IN SCHEMA public GRANT SELECT ON TABLES TO PUBLIC")) + caplog.set_level("INFO", logger="alembic.runtime.migration") + await db_session.run_sync(upgrade) + assert "Verification evidence audit: 2 accounts require fresh proof" in caplog.text + assert not await db_session.scalar( + text("SELECT has_table_privilege(:role, 'verification_deliveries', 'SELECT')"), + {"role": reader_role}, + ) + + await db_session.execute( + text(f'ALTER DEFAULT PRIVILEGES IN SCHEMA public REVOKE SELECT ON TABLES FROM "{reader_role}"') + ) + await db_session.execute(text("ALTER DEFAULT PRIVILEGES IN SCHEMA public REVOKE SELECT ON TABLES FROM PUBLIC")) + await db_session.execute(text(f'DROP ROLE "{reader_role}"')) + rows = ( + await db_session.execute( + text("SELECT primary_email_verified, verified_email, verification_source FROM users ORDER BY primary_email") + ) + ).all() + assert rows == [ + (True, "reader0@example.com", "papyrus_email"), + (False, None, None), + (False, None, None), + (False, None, None), + (True, "reader4@gmail.com", "google"), + ] + assert await db_session.scalar(text("SELECT count(*) FROM users")) == 5 + assert ( + await db_session.scalar( + text("SELECT count(*) FROM password_credentials WHERE password_hash = 'unchanged-hash'") + ) + == 5 + ) + assert await db_session.scalar(text("SELECT count(*) FROM user_identities")) == 3 + assert await db_session.scalar(text("SELECT count(*) FROM auth_sessions WHERE revoked_at IS NULL")) == 2 + assert ( + await db_session.scalar( + text("SELECT verification_subject FROM users WHERE primary_email = 'reader4@gmail.com'"), + ) + == "4" + ) + assert await db_session.scalar(text("SELECT current_cfi FROM books")) == "epub:0:123" + differences = await db_session.run_sync( + lambda session: compare_metadata(MigrationContext.configure(session.connection()), Base.metadata) + ) + assert differences == [] From ae58cac5ae89a5af8208df5b8a19470b94a242d7 Mon Sep 17 00:00:00 2001 From: Karolis Strazdas Date: Fri, 6 Nov 2026 22:29:58 +0200 Subject: [PATCH 2/2] Simplify verification state and normalize delivery metadata --- ...2f3456789ab_simplify_verification_state.py | 57 ++++++ papyrus/models/auth.py | 3 - papyrus/models/user.py | 33 ++-- papyrus/services/account_deletion.py | 14 +- papyrus/services/account_notifications.py | 2 +- papyrus/services/auth/_core.py | 14 +- papyrus/services/auth/email_flows.py | 36 +++- papyrus/services/auth/google.py | 9 +- papyrus/services/auth/sessions.py | 1 - papyrus/services/auth/verification.py | 61 +++--- tests/api/routes/test_auth.py | 4 +- tests/api/routes/test_bookmark_sync.py | 6 +- tests/api/routes/test_library_sync.py | 3 +- tests/api/routes/test_media.py | 3 +- tests/api/routes/test_powersync_sandbox.py | 6 +- tests/api/routes/test_rate_limits.py | 10 +- tests/api/routes/test_sync.py | 3 +- tests/api/routes/test_verification.py | 6 +- tests/conftest.py | 6 +- tests/integration/test_auth_smoke.py | 4 +- tests/services/test_account_notifications.py | 7 +- tests/services/test_auth.py | 10 +- tests/services/test_powersync_demo.py | 3 +- tests/services/test_sync.py | 3 +- tests/services/test_users.py | 3 +- tests/services/test_verification.py | 185 +++++++++++++++++- tests/test_models.py | 7 +- tests/test_reading_timer_migration.py | 3 +- tests/test_tracking_migration.py | 3 +- tests/test_verification_migration.py | 83 +++++++- 30 files changed, 464 insertions(+), 124 deletions(-) create mode 100644 alembic/versions/d2f3456789ab_simplify_verification_state.py diff --git a/alembic/versions/d2f3456789ab_simplify_verification_state.py b/alembic/versions/d2f3456789ab_simplify_verification_state.py new file mode 100644 index 0000000..4a28351 --- /dev/null +++ b/alembic/versions/d2f3456789ab_simplify_verification_state.py @@ -0,0 +1,57 @@ +"""Keep one account verification state and derive outbox metadata from its token. + +Retain the preceding revision so already-applied review/staging databases can +upgrade without losing evidence, pending deliveries, accounts or library data. +""" + +from collections.abc import Sequence + +from alembic import op + +revision: str = "d2f3456789ab" +down_revision: str | Sequence[str] | None = "d1e2f3456789" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + op.execute( + """ + UPDATE users SET verified_at = NULL, verification_source = NULL + WHERE ( + primary_email_verified AND primary_email IS NOT NULL + AND verified_email = primary_email AND verified_at IS NOT NULL + AND verification_source IN ('papyrus_email', 'google') + AND (verification_source <> 'google' OR verification_subject IS NOT NULL) + ) IS NOT TRUE + """, + ) + + op.execute( + "UPDATE auth_sessions SET revoked_at = CURRENT_TIMESTAMP WHERE revoked_at IS NULL " + "AND user_id IN (SELECT user_id FROM users WHERE verified_at IS NULL)", + ) + + op.alter_column("users", "verified_at", new_column_name="email_verified_at") + op.alter_column("users", "verification_source", new_column_name="email_verification_source") + + for column in ("primary_email_verified", "verified_email", "verification_subject"): + op.drop_column("users", column) + + op.create_check_constraint( + "ck_users_email_verification", + "users", + "(email_verified_at IS NULL AND email_verification_source IS NULL) OR " + "(primary_email IS NOT NULL AND email_verified_at IS NOT NULL " + "AND email_verification_source IS NOT NULL " + "AND email_verification_source IN ('papyrus_email', 'google'))", + ) + + op.drop_index("ix_verification_deliveries_user_id", table_name="verification_deliveries") + + for column in ("user_id", "recipient", "expires_at"): + op.drop_column("verification_deliveries", column) + + +def downgrade() -> None: + raise RuntimeError("Retain verification enforcement and pending deliveries. Fix forward.") diff --git a/papyrus/models/auth.py b/papyrus/models/auth.py index 37184ac..756822c 100644 --- a/papyrus/models/auth.py +++ b/papyrus/models/auth.py @@ -159,9 +159,6 @@ class VerificationDelivery(Base): primary_key=True, ) - user_id: Mapped[UUID] = mapped_column(ForeignKey("users.user_id", ondelete="CASCADE"), nullable=False, index=True) - recipient: Mapped[str] = mapped_column(String(255), nullable=False) token: Mapped[str] = mapped_column(Text, nullable=False) attempts: Mapped[int] = mapped_column(Integer, nullable=False, default=0, server_default=text("0")) next_attempt_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False, index=True) - expires_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False) diff --git a/papyrus/models/user.py b/papyrus/models/user.py index 23d84f9..8bf9770 100644 --- a/papyrus/models/user.py +++ b/papyrus/models/user.py @@ -5,7 +5,7 @@ from datetime import datetime from uuid import UUID, uuid4 -from sqlalchemy import Boolean, DateTime, String, Uuid, func, text +from sqlalchemy import CheckConstraint, DateTime, String, Uuid, func from sqlalchemy.orm import Mapped, mapped_column, relationship from papyrus.core.database import Base @@ -15,23 +15,23 @@ class User(Base): """Persisted user account and profile state.""" __tablename__ = "users" + __table_args__ = ( + CheckConstraint( + "(email_verified_at IS NULL AND email_verification_source IS NULL) OR " + "(primary_email IS NOT NULL AND email_verified_at IS NOT NULL " + "AND email_verification_source IS NOT NULL " + "AND email_verification_source IN ('papyrus_email', 'google'))", + name="ck_users_email_verification", + ), + ) user_id: Mapped[UUID] = mapped_column(Uuid, primary_key=True, default=uuid4) display_name: Mapped[str] = mapped_column(String(100), nullable=False) avatar_url: Mapped[str | None] = mapped_column(String(2048), nullable=True) primary_email: Mapped[str | None] = mapped_column(String(255), unique=True, nullable=True) - primary_email_verified: Mapped[bool] = mapped_column( - Boolean, - nullable=False, - default=False, - server_default=text("false"), - ) - - verified_email: Mapped[str | None] = mapped_column(String(255), nullable=True) - verification_source: Mapped[str | None] = mapped_column(String(32), nullable=True) - verification_subject: Mapped[str | None] = mapped_column(String(255), nullable=True) - verified_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + email_verified_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + email_verification_source: Mapped[str | None] = mapped_column(String(32), nullable=True) created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False, server_default=func.now()) last_login_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) @@ -62,14 +62,7 @@ def email(self) -> str | None: @property def email_verified(self) -> bool: - return ( - self.primary_email_verified - and self.primary_email is not None - and self.verified_email == self.primary_email - and self.verification_source in {"papyrus_email", "google"} - and self.verified_at is not None - and (self.verification_source != "google" or self.verification_subject is not None) - ) + return self.email_verified_at is not None from papyrus.models.auth import AuthSession, PasswordCredential, UserIdentity # noqa: E402 diff --git a/papyrus/services/account_deletion.py b/papyrus/services/account_deletion.py index 250bd38..377bd40 100644 --- a/papyrus/services/account_deletion.py +++ b/papyrus/services/account_deletion.py @@ -9,7 +9,7 @@ from typing import NoReturn from uuid import UUID -from sqlalchemy import delete, select, update +from sqlalchemy import delete, select from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker from papyrus.config import get_settings @@ -19,14 +19,12 @@ AccountMediaCleanup, AuthExchangeCode, AuthSession, - EmailActionToken, PasswordCredential, User, - VerificationDelivery, ) from papyrus.schemas.account_deletion import AccountDeletionAuthorization, AccountDeletionStatus, AccountPasswordRequest from papyrus.services.account_notifications import deliver_pending_notifications, queue_account_notification -from papyrus.services.auth._core import _create_session_for_user +from papyrus.services.auth._core import _create_session_for_user, _revoke_credentials from papyrus.services.auth.types import AuthResult from papyrus.services.media import media_root @@ -165,14 +163,6 @@ async def exchange_google(session: AsyncSession, code: str, verifier: str) -> Ac return await _authorize(session, user) -async def _revoke_credentials(session: AsyncSession, user_id: UUID) -> None: - now = _now() - await session.execute(delete(VerificationDelivery).where(VerificationDelivery.user_id == user_id)) - await session.execute(update(AuthSession).where(AuthSession.user_id == user_id).values(revoked_at=now)) - await session.execute(update(AuthExchangeCode).where(AuthExchangeCode.user_id == user_id).values(used_at=now)) - await session.execute(update(EmailActionToken).where(EmailActionToken.user_id == user_id).values(used_at=now)) - - async def schedule_deletion(session: AsyncSession, token: str) -> AccountDeletionStatus: user, authorization = await _lock_authorization(session, token, AUTHORIZATION_PURPOSE) return await _schedule_deletion(session, user, authorization.created_at) diff --git a/papyrus/services/account_notifications.py b/papyrus/services/account_notifications.py index 3967990..81fc45c 100644 --- a/papyrus/services/account_notifications.py +++ b/papyrus/services/account_notifications.py @@ -107,7 +107,7 @@ def render_account_email(display_name: str, deletion_due_at: datetime | None) -> async def queue_account_notification(session: AsyncSession, user: User) -> None: """Keep a notice for each committed lifecycle event, including rapid deletion and recovery.""" - if not user.primary_email or not user.primary_email_verified: + if not user.primary_email or not user.email_verified: return content = render_account_email(user.display_name, user.deletion_due_at) diff --git a/papyrus/services/auth/_core.py b/papyrus/services/auth/_core.py index 447e5f1..a9ddcea 100644 --- a/papyrus/services/auth/_core.py +++ b/papyrus/services/auth/_core.py @@ -6,7 +6,7 @@ from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit from uuid import UUID -from sqlalchemy import select +from sqlalchemy import delete, select, update from sqlalchemy.ext.asyncio import AsyncSession from papyrus.config import get_settings @@ -16,7 +16,7 @@ generate_opaque_token, hash_opaque_token, ) -from papyrus.models import AuthExchangeCode, AuthSession, EmailActionToken, User +from papyrus.models import AuthExchangeCode, AuthSession, EmailActionToken, User, VerificationDelivery from papyrus.services.auth.types import AuthResult, GoogleIdentity from papyrus.services.email import render_email_html @@ -116,6 +116,16 @@ async def _revoke_user_sessions(session: AsyncSession, user_id: UUID) -> None: auth_session.revoked_at = _now() +async def _revoke_credentials(session: AsyncSession, user_id: UUID) -> None: + """Invalidate credentials under the owner's lock without discarding durable account data.""" + tokens = select(EmailActionToken.token_id).where(EmailActionToken.user_id == user_id) + now = _now() + await session.execute(delete(VerificationDelivery).where(VerificationDelivery.token_id.in_(tokens))) + await session.execute(update(AuthSession).where(AuthSession.user_id == user_id).values(revoked_at=now)) + await session.execute(update(AuthExchangeCode).where(AuthExchangeCode.user_id == user_id).values(used_at=now)) + await session.execute(update(EmailActionToken).where(EmailActionToken.user_id == user_id).values(used_at=now)) + + async def _create_exchange_code( session: AsyncSession, *, diff --git a/papyrus/services/auth/email_flows.py b/papyrus/services/auth/email_flows.py index 3df80f6..731e6b8 100644 --- a/papyrus/services/auth/email_flows.py +++ b/papyrus/services/auth/email_flows.py @@ -2,12 +2,19 @@ from __future__ import annotations +from uuid import UUID + from sqlalchemy import select from sqlalchemy.exc import SQLAlchemyError from sqlalchemy.ext.asyncio import AsyncSession from papyrus.config import get_settings -from papyrus.core.exceptions import ValidationError, VerificationTokenInvalidError, VerificationUnavailableError +from papyrus.core.exceptions import ( + NotFoundError, + ValidationError, + VerificationTokenInvalidError, + VerificationUnavailableError, +) from papyrus.core.security import hash_password from papyrus.models import PasswordCredential, User from papyrus.services import email as email_service @@ -15,10 +22,12 @@ _consume_email_action_token, _get_user_by_email, _issue_email_action_token, + _lock_credential_owner, _normalize_email, _now, _password_reset_email_body, _password_reset_email_html, + _revoke_credentials, _revoke_user_sessions, ) from papyrus.services.auth.types import EMAIL_VERIFICATION_ACTION, PASSWORD_RESET_ACTION @@ -31,6 +40,31 @@ ) +async def change_primary_email(session: AsyncSession, user_id: UUID, email: str | None) -> User: + """Change an existing address and invalidate its evidence and credentials atomically. + + The caller commits the transaction. Initial account creation may assign the + address directly because it has no prior evidence or credentials to invalidate. + """ + await _lock_credential_owner(session, user_id) + user = await session.get(User, user_id) + + if user is None: + raise NotFoundError("User not found") + + normalized_email = _normalize_email(email) if email is not None else None + + if user.primary_email == normalized_email: + return user + + user.primary_email = normalized_email + user.email_verified_at = None + user.email_verification_source = None + await _revoke_credentials(session, user.user_id) + await session.flush() + return user + + async def resend_verification_email(session: AsyncSession, email: str) -> str: require_delivery_configuration() diff --git a/papyrus/services/auth/google.py b/papyrus/services/auth/google.py index bfb75a9..98c993d 100644 --- a/papyrus/services/auth/google.py +++ b/papyrus/services/auth/google.py @@ -27,6 +27,7 @@ _normalize_email, _now, ) +from papyrus.services.auth.email_flows import change_primary_email from papyrus.services.auth.types import GOOGLE_PROVIDER, GoogleIdentity from papyrus.services.auth.verification import ( cancel_verification_deliveries, @@ -261,7 +262,7 @@ async def _resolve_google_login_user(session: AsyncSession, identity: GoogleIden if await _get_user_by_email(session, identity.email) is not None: return None, "account_exists" - user.primary_email = identity.email + user = await change_primary_email(session, user.user_id, identity.email) existing_identity.last_used_at = _now() existing_identity.email_at_provider = identity.email @@ -278,7 +279,6 @@ async def _resolve_google_login_user(session: AsyncSession, identity: GoogleIden display_name=_default_display_name(identity, identity.email), avatar_url=identity.avatar_url, primary_email=identity.email, - primary_email_verified=False, ) session.add(user) @@ -297,7 +297,7 @@ async def _resolve_google_login_user(session: AsyncSession, identity: GoogleIden ) if identity.email == user.primary_email and _google_is_authoritative(identity) and not user.email_verified: - record_verification(user, "google", identity.subject) + record_verification(user, "google") await cancel_verification_deliveries(session, user) if not user.email_verified: @@ -461,8 +461,7 @@ async def complete_google_link(session: AsyncSession, user_id: UUID, code: str) ) if user.primary_email is None and exchange_code.email_at_provider is not None: - user.primary_email = exchange_code.email_at_provider - user.primary_email_verified = False + user = await change_primary_email(session, user.user_id, exchange_code.email_at_provider) if user.avatar_url is None and exchange_code.avatar_url is not None: user.avatar_url = exchange_code.avatar_url diff --git a/papyrus/services/auth/sessions.py b/papyrus/services/auth/sessions.py index 3a7ce2d..e570006 100644 --- a/papyrus/services/auth/sessions.py +++ b/papyrus/services/auth/sessions.py @@ -61,7 +61,6 @@ async def register_user( user = User( display_name=request.display_name, primary_email=normalized_email, - primary_email_verified=False, ) session.add(user) diff --git a/papyrus/services/auth/verification.py b/papyrus/services/auth/verification.py index eea6690..b774143 100644 --- a/papyrus/services/auth/verification.py +++ b/papyrus/services/auth/verification.py @@ -4,6 +4,7 @@ import logging import math from datetime import timedelta +from typing import Literal from urllib.parse import urlsplit from sqlalchemy import delete, select, update @@ -41,12 +42,12 @@ def accepts_verification(user: User) -> bool: return user.disabled_at is None -def record_verification(user: User, source: str, subject: str | None = None) -> None: - user.primary_email_verified = True - user.verified_email = user.primary_email - user.verification_source = source - user.verification_subject = subject - user.verified_at = _now() +def record_verification(user: User, source: Literal["papyrus_email", "google"]) -> None: + if user.primary_email is None: + raise ValueError("Verification requires a primary email") + + user.email_verification_source = source + user.email_verified_at = _now() async def require_verified(session: AsyncSession, user: User) -> None: @@ -59,7 +60,9 @@ async def verification_required(session: AsyncSession, user: User) -> Verificati remaining = 0 if recipient is not None: - remaining = max(0, math.ceil((recipient.last_queued_at + timedelta(seconds=60) - _now()).total_seconds())) + remaining = max( + 0, math.ceil((recipient.last_queued_at + timedelta(seconds=COOLDOWN_SECONDS) - _now()).total_seconds()) + ) return VerificationRequiredError(user.primary_email, remaining) @@ -104,11 +107,8 @@ async def queue_verification(session: AsyncSession, user: User) -> bool: session.add( VerificationDelivery( token_id=action.token_id, - user_id=user.user_id, - recipient=user.primary_email, token=token, next_attempt_at=now, - expires_at=expires_at, ), ) @@ -117,21 +117,17 @@ async def queue_verification(session: AsyncSession, user: User) -> bool: async def cancel_verification_deliveries(session: AsyncSession, user: User) -> None: - await session.execute( - delete(VerificationDelivery).where( - VerificationDelivery.user_id == user.user_id, - VerificationDelivery.recipient == user.primary_email, - ), + tokens = select(EmailActionToken.token_id).where( + EmailActionToken.user_id == user.user_id, + EmailActionToken.email == user.primary_email, + EmailActionToken.action_type == EMAIL_VERIFICATION_ACTION, ) + await session.execute(delete(VerificationDelivery).where(VerificationDelivery.token_id.in_(tokens))) + await session.execute( update(EmailActionToken) - .where( - EmailActionToken.user_id == user.user_id, - EmailActionToken.email == user.primary_email, - EmailActionToken.action_type == EMAIL_VERIFICATION_ACTION, - EmailActionToken.used_at.is_(None), - ) + .where(EmailActionToken.token_id.in_(tokens), EmailActionToken.used_at.is_(None)) .values(used_at=_now()), ) @@ -153,7 +149,7 @@ async def deliver_pending_verifications(session: AsyncSession) -> int: await session.execute( delete(VerificationDelivery).where( - (VerificationDelivery.expires_at <= _now()) | VerificationDelivery.token_id.not_in(valid_token), + VerificationDelivery.token_id.not_in(valid_token), ), ) @@ -175,28 +171,33 @@ async def deliver_pending_verifications(session: AsyncSession) -> int: for _ in range(100): now = _now() - job = await session.scalar( - select(VerificationDelivery) - .where(VerificationDelivery.next_attempt_at <= now, VerificationDelivery.expires_at > now) + result = await session.execute( + select(VerificationDelivery, EmailActionToken) + .join(EmailActionToken, EmailActionToken.token_id == VerificationDelivery.token_id) + .where(VerificationDelivery.next_attempt_at <= now, EmailActionToken.expires_at > now) .order_by(VerificationDelivery.next_attempt_at) - .with_for_update(skip_locked=True) + .with_for_update(of=VerificationDelivery, skip_locked=True) .limit(1), ) - if job is None: + delivery = result.one_or_none() + + if delivery is None: await session.rollback() break + job, action = delivery + assert action.email is not None job.attempts += 1 - job.next_attempt_at = min(now + timedelta(minutes=5), job.expires_at) + job.next_attempt_at = min(now + timedelta(minutes=5), action.expires_at) await session.commit() url = _build_app_url("/verify-email", {"token": job.token}) body = f"Verify your Papyrus email address.\n\n{url}\n\nThis link expires 24 hours after it was requested." try: - await asyncio.to_thread(email.send_email, job.recipient, "Verify your Papyrus email address", body) + await asyncio.to_thread(email.send_email, action.email, "Verify your Papyrus email address", body) except Exception: - retry_at = min(_now() + timedelta(seconds=min(60 * 2 ** min(job.attempts - 1, 6), 3600)), job.expires_at) + retry_at = min(_now() + timedelta(seconds=min(60 * 2 ** min(job.attempts - 1, 6), 3600)), action.expires_at) await session.execute( update(VerificationDelivery) diff --git a/tests/api/routes/test_auth.py b/tests/api/routes/test_auth.py index 41c48c3..c536848 100644 --- a/tests/api/routes/test_auth.py +++ b/tests/api/routes/test_auth.py @@ -142,7 +142,9 @@ async def test_register_duplicate_email_returns_conflict(client: AsyncClient): async def verify_and_login(client, db_session, email="test@example.com"): - delivery = await db_session.scalar(select(VerificationDelivery).where(VerificationDelivery.recipient == email)) + delivery = await db_session.scalar( + select(VerificationDelivery).join(EmailActionToken).where(EmailActionToken.email == email) + ) assert (await client.post("/v1/auth/verify-email", json={"token": delivery.token})).status_code == 200 response = await client.post("/v1/auth/login", json={"email": email, "password": "SecureP@ss123"}) assert response.status_code == 200 diff --git a/tests/api/routes/test_bookmark_sync.py b/tests/api/routes/test_bookmark_sync.py index 691c969..a5b0f08 100644 --- a/tests/api/routes/test_bookmark_sync.py +++ b/tests/api/routes/test_bookmark_sync.py @@ -154,7 +154,8 @@ async def test_bookmark_rejects_foreign_book_and_rolls_back(client, auth_headers owner = User( display_name="Other", primary_email="bookmark-owner@example.com", - primary_email_verified=True, + email_verified_at=datetime.now(UTC), + email_verification_source="papyrus_email", last_login_at=datetime.now(UTC), ) @@ -188,7 +189,8 @@ async def test_bookmark_rejects_foreign_entity_mutations(client, auth_headers, d owner = User( display_name="Other", primary_email="bookmark-entity@example.com", - primary_email_verified=True, + email_verified_at=datetime.now(UTC), + email_verification_source="papyrus_email", last_login_at=datetime.now(UTC), ) diff --git a/tests/api/routes/test_library_sync.py b/tests/api/routes/test_library_sync.py index 4dd41ca..1da870a 100644 --- a/tests/api/routes/test_library_sync.py +++ b/tests/api/routes/test_library_sync.py @@ -180,7 +180,8 @@ async def test_foreign_reference_rolls_back_mixed_batch(client, auth_headers, db other = User( display_name="Other", primary_email="other-library@example.com", - primary_email_verified=True, + email_verified_at=datetime.now(UTC), + email_verification_source="papyrus_email", last_login_at=datetime.now(UTC), ) diff --git a/tests/api/routes/test_media.py b/tests/api/routes/test_media.py index 0efc0fd..8132db5 100644 --- a/tests/api/routes/test_media.py +++ b/tests/api/routes/test_media.py @@ -167,7 +167,8 @@ async def test_upload_rejects_cross_user_book( other_user = User( display_name="Other User", primary_email="other-media@example.com", - primary_email_verified=True, + email_verified_at=datetime.now(UTC), + email_verification_source="papyrus_email", last_login_at=datetime.now(UTC), ) diff --git a/tests/api/routes/test_powersync_sandbox.py b/tests/api/routes/test_powersync_sandbox.py index 134a3e2..9c199b5 100644 --- a/tests/api/routes/test_powersync_sandbox.py +++ b/tests/api/routes/test_powersync_sandbox.py @@ -179,7 +179,8 @@ async def test_powersync_demo_items_only_lists_owned_rows( other_user = User( display_name="Other User", primary_email="other@example.com", - primary_email_verified=True, + email_verified_at=datetime.now(UTC), + email_verification_source="papyrus_email", last_login_at=datetime.now(UTC), ) @@ -265,7 +266,8 @@ async def test_powersync_upload_rejects_mutating_other_users_rows( other_user = User( display_name="Other User", primary_email="other-owner@example.com", - primary_email_verified=True, + email_verified_at=datetime.now(UTC), + email_verification_source="papyrus_email", last_login_at=datetime.now(UTC), ) diff --git a/tests/api/routes/test_rate_limits.py b/tests/api/routes/test_rate_limits.py index 328ad3c..433d63c 100644 --- a/tests/api/routes/test_rate_limits.py +++ b/tests/api/routes/test_rate_limits.py @@ -62,10 +62,12 @@ async def test_upload_limits_are_per_validated_user_across_sessions( monkeypatch.setattr(settings, limit_setting, 1) monkeypatch.setattr(settings, "media_storage_root", str(tmp_path)) monkeypatch.setattr(settings, "cors_origins", ["https://app.papyrus.test"]) - other_user = User(display_name="Other", primary_email="rate-other@example.com", primary_email_verified=True) - other_user.verified_email = other_user.primary_email - other_user.verification_source = "papyrus_email" - other_user.verified_at = datetime.now(UTC) + other_user = User( + display_name="Other", + primary_email="rate-other@example.com", + email_verified_at=datetime.now(UTC), + email_verification_source="papyrus_email", + ) db_session.add(other_user) await db_session.flush() owner_id = UUID(auth_user["user_id"]) diff --git a/tests/api/routes/test_sync.py b/tests/api/routes/test_sync.py index 7d78cbf..5aba53c 100644 --- a/tests/api/routes/test_sync.py +++ b/tests/api/routes/test_sync.py @@ -259,7 +259,8 @@ async def test_powersync_upload_rejects_cross_user_book_mutation( other_user = User( display_name="Other User", primary_email="other-sync@example.com", - primary_email_verified=True, + email_verified_at=datetime.now(UTC), + email_verification_source="papyrus_email", last_login_at=datetime.now(UTC), ) diff --git a/tests/api/routes/test_verification.py b/tests/api/routes/test_verification.py index 3e6658f..71c0e4c 100644 --- a/tests/api/routes/test_verification.py +++ b/tests/api/routes/test_verification.py @@ -30,7 +30,8 @@ def mail_configuration(monkeypatch): ) async def test_legacy_jwt_cannot_access_application(client, auth_user, auth_headers, db_session, path, method, body): user = await db_session.get(User, UUID(auth_user["user_id"])) - user.verified_email = "old@example.com" + user.email_verified_at = None + user.email_verification_source = None await db_session.commit() response = await client.request(method, path, headers=auth_headers, json=body) assert response.status_code == 403, response.text @@ -41,7 +42,8 @@ async def test_legacy_jwt_cannot_access_application(client, auth_user, auth_head async def test_logout_remains_available_for_unverified_session(client, auth_user, auth_headers, db_session): user = await db_session.get(User, UUID(auth_user["user_id"])) - user.verified_email = None + user.email_verified_at = None + user.email_verification_source = None await db_session.commit() assert (await client.post("/v1/auth/logout", headers=auth_headers)).status_code == 204 await db_session.refresh(user) diff --git a/tests/conftest.py b/tests/conftest.py index 8da6b5e..ff59c67 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -302,10 +302,8 @@ async def auth_user( user_id=user_uuid, display_name="Example User", primary_email="user@example.com", - primary_email_verified=True, - verified_email="user@example.com", - verification_source="papyrus_email", - verified_at=datetime.now(UTC), + email_verified_at=datetime.now(UTC), + email_verification_source="papyrus_email", last_login_at=datetime.now(UTC), ) diff --git a/tests/integration/test_auth_smoke.py b/tests/integration/test_auth_smoke.py index f981332..eb90d0d 100644 --- a/tests/integration/test_auth_smoke.py +++ b/tests/integration/test_auth_smoke.py @@ -3,6 +3,7 @@ from __future__ import annotations import os +from datetime import UTC, datetime import httpx import pytest @@ -41,7 +42,8 @@ async def test_smtp_password_reset_smoke( User( display_name="SMTP Smoke", primary_email=recipient, - primary_email_verified=True, + email_verified_at=datetime.now(UTC), + email_verification_source="papyrus_email", ) ) diff --git a/tests/services/test_account_notifications.py b/tests/services/test_account_notifications.py index 87f5ac5..c38fdea 100644 --- a/tests/services/test_account_notifications.py +++ b/tests/services/test_account_notifications.py @@ -164,10 +164,11 @@ async def test_notices_roll_back_expire_and_cascade_with_account(db_session, aut assert await db_session.scalar(select(func.count()).select_from(AccountNotification)) == 0 -@pytest.mark.parametrize("verified,email_address", [(False, "reader@example.test"), (True, None)]) -async def test_only_verified_contact_addresses_receive_notices(db_session, auth_user, verified, email_address): +@pytest.mark.parametrize("email_address", ["reader@example.test", None]) +async def test_only_verified_contact_addresses_receive_notices(db_session, auth_user, email_address): user = await db_session.get(User, UUID(auth_user["user_id"])) - user.primary_email_verified = verified + user.email_verified_at = None + user.email_verification_source = None user.primary_email = email_address await account_notifications.queue_account_notification(db_session, user) await db_session.commit() diff --git a/tests/services/test_auth.py b/tests/services/test_auth.py index 85f2e14..ef9e5bc 100644 --- a/tests/services/test_auth.py +++ b/tests/services/test_auth.py @@ -58,7 +58,8 @@ async def _create_user_with_password( user = User( display_name=display_name, primary_email=email, - primary_email_verified=True, + email_verified_at=datetime.now(UTC), + email_verification_source="papyrus_email", disabled_at=datetime.now(UTC) if disabled else None, ) @@ -82,7 +83,9 @@ def verification_configuration(monkeypatch): async def register_verified(session, request, user_agent): await auth_service.register_user(session, request, user_agent) - delivery = await session.scalar(select(VerificationDelivery).where(VerificationDelivery.recipient == request.email)) + delivery = await session.scalar( + select(VerificationDelivery).join(EmailActionToken).where(EmailActionToken.email == request.email) + ) await auth_service.verify_email_token(session, delivery.token) return await auth_service.login_user( session, LoginRequest(email=request.email, password=request.password), user_agent @@ -270,7 +273,8 @@ async def test_google_login_reuses_existing_identity( user = User( display_name="Google User", primary_email="google@example.com", - primary_email_verified=True, + email_verified_at=datetime.now(UTC), + email_verification_source="papyrus_email", ) record_verification(user, "papyrus_email") diff --git a/tests/services/test_powersync_demo.py b/tests/services/test_powersync_demo.py index 8bf3ed8..fad8e54 100644 --- a/tests/services/test_powersync_demo.py +++ b/tests/services/test_powersync_demo.py @@ -18,7 +18,8 @@ async def _create_user(session: AsyncSession, email: str) -> User: user = User( display_name=email.split("@", 1)[0], primary_email=email, - primary_email_verified=True, + email_verified_at=datetime.now(UTC), + email_verification_source="papyrus_email", last_login_at=datetime.now(UTC), ) diff --git a/tests/services/test_sync.py b/tests/services/test_sync.py index 8dae7a5..597ce09 100644 --- a/tests/services/test_sync.py +++ b/tests/services/test_sync.py @@ -18,7 +18,8 @@ async def _create_user(session: AsyncSession, email: str) -> User: user = User( display_name=email.split("@", 1)[0], primary_email=email, - primary_email_verified=True, + email_verified_at=datetime.now(UTC), + email_verification_source="papyrus_email", last_login_at=datetime.now(UTC), ) diff --git a/tests/services/test_users.py b/tests/services/test_users.py index bc296e8..0166cff 100644 --- a/tests/services/test_users.py +++ b/tests/services/test_users.py @@ -23,7 +23,8 @@ async def _seed_user_with_session( user = User( display_name="Example User", primary_email=email, - primary_email_verified=True, + email_verified_at=datetime.now(UTC), + email_verification_source="papyrus_email", last_login_at=datetime.now(UTC), ) diff --git a/tests/services/test_verification.py b/tests/services/test_verification.py index b84457d..2d7f700 100644 --- a/tests/services/test_verification.py +++ b/tests/services/test_verification.py @@ -64,7 +64,7 @@ async def test_pending_denial_and_explicit_verification(test_session_maker): assert await session.scalar(select(AuthSession)) is None result = await auth.login_user(session, LoginRequest(email=user.primary_email, password="SecureP@ss123"), None) assert result.user.email_verified - assert result.user.verification_source == "papyrus_email" + assert result.user.email_verification_source == "papyrus_email" assert await session.scalar(select(VerificationDelivery)) is None @@ -158,12 +158,14 @@ def fail(*args, **kwargs): async with test_session_maker() as session: await register(session) job = await session.scalar(select(VerificationDelivery)) - original_token, original_expiry = job.token, job.expires_at + action = await session.get(EmailActionToken, job.token_id) + original_token, original_expiry = job.token, action.expires_at assert await verification.deliver_pending_verifications(session) == 0 await session.refresh(job) + await session.refresh(action) assert job.attempts == 1 assert job.next_attempt_at > datetime.now(UTC) - assert job.token == original_token and job.expires_at == original_expiry + assert job.token == original_token and action.expires_at == original_expiry assert original_token not in caplog.text assert "SMTP secret" not in caplog.text job.next_attempt_at = datetime.now(UTC) - timedelta(seconds=1) @@ -181,7 +183,7 @@ async def test_expired_delivery_removed_without_smtp(test_session_maker, monkeyp async with test_session_maker() as session: await register(session) - await session.execute(update(VerificationDelivery).values(expires_at=datetime.now(UTC) - timedelta(seconds=1))) + await session.execute(update(EmailActionToken).values(expires_at=datetime.now(UTC) - timedelta(seconds=1))) await session.commit() assert await verification.deliver_pending_verifications(session) == 0 assert await session.scalar(select(VerificationDelivery)) is None @@ -237,7 +239,7 @@ async def test_google_authority_and_pending_handoff(test_session_maker, monkeypa if eligible: result = await auth.exchange_login_code(session, OAuthExchangeRequest(code=code), None) assert result.user.email_verified - assert result.user.verification_subject == "subject" + assert stored.provider_subject == "subject" else: job = await session.scalar(select(VerificationDelivery)) await auth.verify_email_token(session, job.token) @@ -284,7 +286,8 @@ async def test_google_missing_email_and_delivery_outage(test_session_maker, monk async def test_legacy_refresh_revocation_commits_and_rotation_serializes(test_session_maker, auth_user): async with test_session_maker() as session: user = await session.scalar(select(User)) - user.verified_email = None + user.email_verified_at = None + user.email_verification_source = None await session.commit() with pytest.raises(VerificationRequiredError): @@ -321,7 +324,8 @@ async def test_recovery_requires_evidence_without_consuming_authorization(test_s ) await account_deletion.schedule_deletion(session, proof.authorization_token) user = await session.scalar(select(User)) - user.verified_email = None + user.email_verified_at = None + user.email_verification_source = None await session.commit() proof = await account_deletion.authenticate_password( session, AccountPasswordRequest(email="user@example.com", password=auth_user["password"]) @@ -386,7 +390,8 @@ async def test_linked_google_preserves_owner_and_primary_email( assert await session.scalar(select(AuthExchangeCode)) is None elif expected == "pending": job = await session.scalar(select(VerificationDelivery)) - assert job.recipient == primary + action = await session.get(EmailActionToken, job.token_id) + assert action.email == primary assert stored.primary_email == primary assert not stored.email_verified else: @@ -437,3 +442,167 @@ async def test_new_account_cannot_commit_without_its_delivery_during_recipient_c assert await session.scalar(select(User)) is None assert await session.scalar(select(EmailActionToken)) is None assert await session.scalar(select(VerificationDelivery)) is None + + +async def test_email_change_clears_evidence_and_revokes_credentials_without_touching_library( + test_session_maker, auth_user +): + from uuid import UUID + + from papyrus.models import SyncBook + from papyrus.services.auth import _core + from papyrus.services.auth.email_flows import change_primary_email + + user_id = UUID(auth_user["user_id"]) + + async with test_session_maker() as session: + session.add(SyncBook(owner_user_id=user_id, title="Durable library", current_cfi="epub:0:42")) + reset_token = await _core._issue_email_action_token(session, user_id, auth.PASSWORD_RESET_ACTION, 60) + handoff = await _core._create_exchange_code(session, user_id=user_id, purpose="login", redirect_uri="") + await session.commit() + user = await change_primary_email(session, user_id, "Replacement@Example.com") + await session.commit() + assert user.primary_email == "replacement@example.com" + assert user.email_verified_at is None + assert user.email_verification_source is None + assert not user.email_verified + assert await session.scalar(select(AuthSession).where(AuthSession.revoked_at.is_(None))) is None + assert await session.scalar(select(EmailActionToken).where(EmailActionToken.used_at.is_(None))) is None + assert await session.scalar(select(AuthExchangeCode).where(AuthExchangeCode.used_at.is_(None))) is None + assert ( + await session.scalar(select(SyncBook.current_cfi).where(SyncBook.owner_user_id == user_id)) == "epub:0:42" + ) + + with pytest.raises(AppError): + await auth.reset_password(session, reset_token, "NewSecureP@ss123") + + with pytest.raises(AppError): + await auth.exchange_login_code(session, OAuthExchangeRequest(code=handoff), None) + + +async def test_email_change_away_and_back_cannot_reactivate_old_verification(test_session_maker): + from papyrus.services.auth.email_flows import change_primary_email + + async with test_session_maker() as session: + user = await register(session) + user_id = user.user_id + token = await session.scalar(select(VerificationDelivery.token)) + await change_primary_email(session, user_id, "replacement@example.com") + await session.commit() + assert await session.scalar(select(VerificationDelivery)) is None + await change_primary_email(session, user_id, "reader@example.com") + await session.commit() + + with pytest.raises(VerificationTokenInvalidError): + await auth.verify_email_token(session, token) + + assert not user.email_verified + + +async def test_email_change_noop_and_rollback_preserve_verified_access(test_session_maker, auth_user): + from uuid import UUID + + from papyrus.services.auth.email_flows import change_primary_email + + user_id = UUID(auth_user["user_id"]) + + async with test_session_maker() as session: + user = await change_primary_email(session, user_id, " USER@EXAMPLE.COM ") + await session.commit() + assert user.email_verified + assert await session.scalar(select(AuthSession).where(AuthSession.revoked_at.is_(None))) is not None + await change_primary_email(session, user_id, "replacement@example.com") + await session.rollback() + + async with test_session_maker() as session: + user = await session.get(User, user_id) + assert user.primary_email == "user@example.com" + assert user.email_verified + assert await session.scalar(select(AuthSession).where(AuthSession.revoked_at.is_(None))) is not None + + +async def test_email_change_serializes_with_waiting_verification(test_session_maker, monkeypatch): + from papyrus.services.auth import email_flows + + async with test_session_maker() as session: + user = await register(session) + user_id = user.user_id + token = await session.scalar(select(VerificationDelivery.token)) + + locked = asyncio.Event() + resume = asyncio.Event() + revoke = email_flows._revoke_credentials + + async def pause_before_revocation(session, owner_id): + locked.set() + await resume.wait() + await revoke(session, owner_id) + + monkeypatch.setattr(email_flows, "_revoke_credentials", pause_before_revocation) + + async with test_session_maker() as changing, test_session_maker() as confirming, test_session_maker() as monitor: + changing_pid = await changing.scalar(select(func.pg_backend_pid())) + confirming_pid = await confirming.scalar(select(func.pg_backend_pid())) + + async def change(): + await email_flows.change_primary_email(changing, user_id, "replacement@example.com") + await changing.commit() + + async def confirm(): + with pytest.raises(VerificationTokenInvalidError): + await auth.verify_email_token(confirming, token) + + change_task = asyncio.create_task(change()) + confirm_task = None + + try: + await asyncio.wait_for(locked.wait(), timeout=5) + confirm_task = asyncio.create_task(confirm()) + + async with asyncio.timeout(5): + while changing_pid not in await monitor.scalar(select(func.pg_blocking_pids(confirming_pid))): + await asyncio.sleep(0.01) + + resume.set() + await asyncio.wait_for(asyncio.gather(change_task, confirm_task), timeout=5) + finally: + resume.set() + tasks = [task for task in (change_task, confirm_task) if task is not None] + + for task in tasks: + if not task.done(): + task.cancel() + + await asyncio.gather(*tasks, return_exceptions=True) + + async with test_session_maker() as session: + user = await session.get(User, user_id) + assert user.primary_email == "replacement@example.com" + assert not user.email_verified + assert await session.scalar(select(VerificationDelivery)) is None + + +@pytest.mark.parametrize( + "address,timestamp,source", + [ + (None, True, "papyrus_email"), + ("reader@example.com", True, None), + ("reader@example.com", False, "papyrus_email"), + ("reader@example.com", True, "unknown"), + ], +) +async def test_database_rejects_inconsistent_verification_evidence(test_session_maker, address, timestamp, source): + from sqlalchemy.exc import IntegrityError + + async with test_session_maker() as session: + session.add( + User( + display_name="Reader", + primary_email=address, + email_verified_at=datetime.now(UTC) if timestamp else None, + email_verification_source=source, + ) + ) + + with pytest.raises(IntegrityError): + await session.flush() diff --git a/tests/test_models.py b/tests/test_models.py index 195247b..f93e304 100644 --- a/tests/test_models.py +++ b/tests/test_models.py @@ -92,11 +92,8 @@ def test_auth_models_are_registered_with_metadata() -> None: "display_name", "avatar_url", "primary_email", - "primary_email_verified", - "verified_email", - "verification_source", - "verification_subject", - "verified_at", + "email_verified_at", + "email_verification_source", "created_at", "last_login_at", "disabled_at", diff --git a/tests/test_reading_timer_migration.py b/tests/test_reading_timer_migration.py index c06f0b3..18d80d1 100644 --- a/tests/test_reading_timer_migration.py +++ b/tests/test_reading_timer_migration.py @@ -32,7 +32,8 @@ def migrate(session, fn): user_id=uuid4(), display_name="Reader", primary_email="migration@example.com", - primary_email_verified=True, + email_verified_at=datetime.now(UTC), + email_verification_source="papyrus_email", last_login_at=datetime.now(UTC), ) diff --git a/tests/test_tracking_migration.py b/tests/test_tracking_migration.py index f1b51d0..908b7d7 100644 --- a/tests/test_tracking_migration.py +++ b/tests/test_tracking_migration.py @@ -32,7 +32,8 @@ def migrate(session, fn): user_id=uuid4(), display_name="Reader", primary_email="migration@example.com", - primary_email_verified=True, + email_verified_at=datetime.now(UTC), + email_verification_source="papyrus_email", last_login_at=datetime.now(UTC), ) diff --git a/tests/test_verification_migration.py b/tests/test_verification_migration.py index 489500f..7bb925d 100644 --- a/tests/test_verification_migration.py +++ b/tests/test_verification_migration.py @@ -21,11 +21,18 @@ async def test_verification_migration_preserves_trustworthy_evidence_and_data(db revision = importlib.util.module_from_spec(spec) spec.loader.exec_module(revision) + simplify_path = path.with_name("d2f3456789ab_simplify_verification_state.py") + simplify_spec = importlib.util.spec_from_file_location("simplification_revision", simplify_path) + simplification = importlib.util.module_from_spec(simplify_spec) + simplify_spec.loader.exec_module(simplification) await db_session.execute(text("DROP TABLE verification_deliveries, verification_recipients")) + await db_session.execute(text("ALTER TABLE users DROP CONSTRAINT ck_users_email_verification")) - for column in ("verified_email", "verification_source", "verification_subject", "verified_at"): + for column in ("email_verified_at", "email_verification_source"): await db_session.execute(text(f"ALTER TABLE users DROP COLUMN {column}")) + await db_session.execute(text("ALTER TABLE users ADD COLUMN primary_email_verified boolean NOT NULL DEFAULT false")) + for table in ("user_identities", "auth_exchange_codes"): await db_session.execute(text(f"ALTER TABLE {table} DROP COLUMN hosted_domain")) @@ -92,16 +99,80 @@ def upgrade(session): ) await db_session.execute(text("ALTER DEFAULT PRIVILEGES IN SCHEMA public REVOKE SELECT ON TABLES FROM PUBLIC")) await db_session.execute(text(f'DROP ROLE "{reader_role}"')) + token_id = uuid4() + + await db_session.execute( + text( + "INSERT INTO email_action_tokens (token_id, user_id, email, action_type, token_hash, expires_at) " + "VALUES (:token, :owner, 'reader1@example.com', 'verify_email', 'preserved-hash', now() + interval '1 day')" + ), + {"token": token_id, "owner": identifiers[1]}, + ) + + await db_session.execute( + text( + "INSERT INTO verification_deliveries (token_id, user_id, recipient, token, attempts, next_attempt_at, expires_at) " + "SELECT token_id, user_id, email, 'preserved-secret', 2, now(), expires_at FROM email_action_tokens WHERE token_id = :token" + ), + {"token": token_id}, + ) + + before = ( + await db_session.execute( + text("SELECT user_id, email, expires_at FROM email_action_tokens WHERE token_id = :token"), + {"token": token_id}, + ) + ).one() + + await db_session.execute( + text( + "UPDATE users SET primary_email_verified = true, verified_email = 'stale@example.com', " + "verification_source = 'papyrus_email', verified_at = now() WHERE user_id = :user" + ), + {"user": identifiers[3]}, + ) + + verified_at = await db_session.scalar( + text("SELECT verified_at FROM users WHERE user_id = :user"), + {"user": identifiers[0]}, + ) + + def simplify(session): + with Operations.context(MigrationContext.configure(session.connection())): + simplification.upgrade() + + await db_session.run_sync(simplify) + assert ( + await db_session.scalar( + text("SELECT email_verified_at FROM users WHERE user_id = :user"), + {"user": identifiers[0]}, + ) + == verified_at + ) + after = ( + await db_session.execute( + text("SELECT user_id, email, expires_at FROM email_action_tokens WHERE token_id = :token"), + {"token": token_id}, + ) + ).one() + assert after == before + assert (await db_session.execute(text("SELECT token, attempts FROM verification_deliveries"))).one() == ( + "preserved-secret", + 2, + ) + rows = ( await db_session.execute( - text("SELECT primary_email_verified, verified_email, verification_source FROM users ORDER BY primary_email") + text( + "SELECT email_verified_at IS NOT NULL, primary_email, email_verification_source FROM users ORDER BY primary_email" + ) ) ).all() assert rows == [ (True, "reader0@example.com", "papyrus_email"), - (False, None, None), - (False, None, None), - (False, None, None), + (False, "reader1@example.com", None), + (False, "reader2@example.com", None), + (False, "reader3@example.com", None), (True, "reader4@gmail.com", "google"), ] assert await db_session.scalar(text("SELECT count(*) FROM users")) == 5 @@ -115,7 +186,7 @@ def upgrade(session): assert await db_session.scalar(text("SELECT count(*) FROM auth_sessions WHERE revoked_at IS NULL")) == 2 assert ( await db_session.scalar( - text("SELECT verification_subject FROM users WHERE primary_email = 'reader4@gmail.com'"), + text("SELECT provider_subject FROM user_identities WHERE email_at_provider = 'reader4@gmail.com'"), ) == "4" )