feat: email verification for registration, email change, and invitations

Registration now creates a pending account and sends a 6-digit code via
email; the account activates only after verification. Email changes on
the profile page follow the same flow. Group invitations send a
notification email to the invitee (without revealing their address to
the inviter) containing the invite code and hub link.

Backend: blind HMAC-SHA256 email index for uniqueness without decryption,
mail.py for localhost Postfix delivery, verification endpoints, cleanup
of expired codes and stale pending accounts, startup backfill of
email_hash for existing users.

Frontend: 3-phase register page, inline email change verification on
profile, invite-notify call with status display. All 10 locales updated.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

2 files changed, 263 insertions, 25 deletions

diff --git a/packages/meshbay-hub/src/meshbay_hub/api/groups.py b/packages/meshbay-hub/src/meshbay_hub/api/groups.py
index 49902de..fcf0360 100644
--- a/packages/meshbay-hub/src/meshbay_hub/api/groups.py
+++ b/packages/meshbay-hub/src/meshbay_hub/api/groups.py
@@ -7,7 +7,8 @@ from sqlalchemy import func, or_, select, update
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from meshbay_hub import hub_settings
+from meshbay_hub import hub_settings, mail
+from meshbay_hub.auth import decrypt_email
 from meshbay_hub.api.deps import get_current_user, require_user_scope
 from meshbay_hub.api.netutil import client_ip
 from meshbay_hub.db.engine import get_db
@@ -671,3 +672,53 @@ async def delete_group(
     return {"status": "deleted", "group_id": group_id}
 
 
+class InviteNotifyRequest(BaseModel):
+    username:   str
+    code:       str
+    group_name: str
+
+
+@router.post("/{group_id}/invite-notify")
+async def invite_notify(
+    group_id: str,
+    body: InviteNotifyRequest,
+    current_user: User = Depends(require_user_scope),
+    db: AsyncSession = Depends(get_db),
+):
+    """Send an invitation email to a member who was just invited.
+
+    The invite code was created on the node — the hub only knows about it
+    because the inviter's browser sends it here. The hub looks up the
+    invitee's encrypted email, decrypts it, and sends the notification.
+    The inviter never sees the email address.
+    """
+    group = await db.get(Group, group_id)
+    if not group:
+        raise HTTPException(status_code=404, detail="Group not found")
+    if group.admin_id != current_user.id:
+        raise HTTPException(status_code=403,
+                            detail="Only the group owner can send invitations")
+
+    target = (await db.execute(
+        select(User).where(User.username == body.username))).scalar_one_or_none()
+    if not target:
+        raise HTTPException(status_code=404, detail="User not found")
+
+    email = ""
+    try:
+        email = decrypt_email(target.email) if target.email else ""
+    except Exception:
+        pass
+
+    if not email:
+        return {"status": "no_email"}
+
+    try:
+        mail.send_invite_notification(
+            email, body.code, current_user.username, body.group_name)
+    except Exception:
+        return {"status": "send_failed"}
+
+    return {"status": "sent"}
+
+
diff --git a/packages/meshbay-hub/src/meshbay_hub/api/users.py b/packages/meshbay-hub/src/meshbay_hub/api/users.py
index 7db3fc9..953ad1b 100644
--- a/packages/meshbay-hub/src/meshbay_hub/api/users.py
+++ b/packages/meshbay-hub/src/meshbay_hub/api/users.py
@@ -1,39 +1,40 @@
 """User endpoints — /v1/users/*"""
 
 import base64
-import time
 import logging
+import secrets
+import time
 import uuid
-from datetime import datetime, timezone, timedelta
+from datetime import datetime, timedelta, timezone
 
 from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey
-from fastapi import APIRouter, Depends, HTTPException, Request, status
+from fastapi import APIRouter, Depends, HTTPException, Request
 from pydantic import BaseModel, field_validator
 from sqlalchemy import delete, select, update
 from sqlalchemy.ext.asyncio import AsyncSession
 
+from meshbay_hub import mail
+from meshbay_hub.api.deps import get_current_user, require_user_scope
+from meshbay_hub.api.middleware import limiter
+from meshbay_hub.api.netutil import client_ip
 from meshbay_hub.auth import (
     current_pw_version,
-    decode_access_token,
     decrypt_email,
     encrypt_email,
     generate_refresh_token,
+    hash_email_blind,
     hash_password,
     hash_refresh_token,
-    hub_public_key_pem,
     issue_access_token,
     pw_needs_rehash,
     verify_password,
 )
-from meshbay_hub.api.middleware import limiter
-from meshbay_hub.api.netutil import client_ip
 from meshbay_hub.config import HubConfig
 from meshbay_hub.db.engine import get_db
 from meshbay_hub.db.models import (
-    Group, GroupMember, IPLog, Node, Notification, RefreshToken, User,
-    UserDevice, UserPreference,
+    EmailVerification, Group, GroupMember, IPLog, Node, Notification,
+    RefreshToken, User, UserDevice, UserPreference,
 )
-from meshbay_hub.api.deps import get_current_user, require_user_scope
 
 log = logging.getLogger(__name__)
 
@@ -51,6 +52,13 @@ def _ttl() -> int:
 def _refresh_ttl() -> int:
     return _cfg.jwt.refresh_token_ttl if _cfg else 86400 * 30
 
+VERIFICATION_TTL = 86400  # 24 hours
+VERIFICATION_MAX_ATTEMPTS = 10
+
+
+def _generate_code() -> str:
+    return f"{secrets.randbelow(1_000_000):06d}"
+
 
 # ── Models ────────────────────────────────────────────────────────────────────
 
@@ -108,36 +116,43 @@ async def register(
     request: Request,
     db: AsyncSession = Depends(get_db),
 ):
+    eh = hash_email_blind(body.email)
+
     existing = await db.execute(
         select(User).where(User.username == body.username))
-    if existing.scalar_one_or_none():
+    found = existing.scalar_one_or_none()
+
+    if found:
+        if found.status == "pending" and found.email_hash == eh:
+            # Same person retrying before validation — resend a code
+            await _create_and_send_verification(db, found, body.email, eh)
+            await db.commit()
+            return {"user_id": found.id, "email_verification_required": True}
         raise HTTPException(status_code=409, detail="Username already taken")
 
+    # Email uniqueness (only active or pending accounts)
+    dup = await db.execute(
+        select(User).where(User.email_hash == eh, User.status.in_(["active", "pending"])))
+    if dup.scalar_one_or_none():
+        raise HTTPException(status_code=409, detail="Email already in use")
+
     credential = body.auth_key or body.password
     if not credential:
         raise HTTPException(status_code=400, detail="auth_key or password required")
 
     pw_hash, pw_salt = hash_password(credential)
-    # auth_key → pw_version 3 (password split); raw password → pw_version 2 (legacy)
     pw_ver = current_pw_version() if body.auth_key else 2
     hub_id = _cfg.identity.id if _cfg else "meshbay.org"
     user = User(
         username=body.username,
         email=encrypt_email(body.email),
+        email_hash=eh,
         pw_hash=pw_hash,
         pw_salt=pw_salt,
         pw_version=pw_ver,
         hub_id=hub_id,
     )
     db.add(user)
-    # flush assigns user.id so the log row can be attributed directly.
-    #
-    # Finding M6: this used to insert the row with a NULL user_id and then run
-    #   UPDATE ip_logs SET user_id = <new user> WHERE user_id IS NULL
-    # which claimed *every* unattributed row in the table — failed logins for other
-    # usernames, other registrations racing this one — and stamped them with the
-    # account just created. For logs retained a year to answer legal requests, that
-    # attributed other people's connections to the wrong person.
     await db.flush()
     db.add(IPLog(
         user_id=user.id,
@@ -145,10 +160,85 @@ async def register(
         ip_address=client_ip(request),
         detail=body.username,
     ))
+
+    await _create_and_send_verification(db, user, body.email, eh)
     await db.commit()
-    await db.refresh(user)
 
-    return {"user_id": user.id}
+    return {"user_id": user.id, "email_verification_required": True}
+
+
+async def _create_and_send_verification(
+    db: AsyncSession, user: User, email: str, eh: str,
+) -> None:
+    user.status = "pending"
+    # Invalidate any previous pending verification for this user
+    prev = await db.execute(
+        select(EmailVerification).where(
+            EmailVerification.user_id == user.id,
+            EmailVerification.purpose == "registration",
+            EmailVerification.verified_at.is_(None),
+        ))
+    for old in prev.scalars().all():
+        await db.delete(old)
+
+    code = _generate_code()
+    db.add(EmailVerification(
+        email_hash=eh,
+        code=code,
+        purpose="registration",
+        user_id=user.id,
+        expires_at=datetime.now(timezone.utc) + timedelta(seconds=VERIFICATION_TTL),
+    ))
+    await db.flush()
+    mail.send_verification_code(email, code)
+
+
+class VerifyEmailRequest(BaseModel):
+    email: str
+    code:  str
+
+
+@router.post("/verify-email")
+@limiter.limit("10/minute")
+async def verify_email(
+    body: VerifyEmailRequest,
+    request: Request,
+    db: AsyncSession = Depends(get_db),
+):
+    """Verify a registration email with the code received by mail."""
+    eh = hash_email_blind(body.email)
+    now = datetime.now(timezone.utc)
+
+    result = await db.execute(
+        select(EmailVerification).where(
+            EmailVerification.email_hash == eh,
+            EmailVerification.purpose == "registration",
+            EmailVerification.verified_at.is_(None),
+        ).order_by(EmailVerification.created_at.desc()))
+    verif = result.scalar_one_or_none()
+
+    if not verif:
+        raise HTTPException(status_code=404,
+                            detail="No pending verification for this email")
+
+    if verif.expires_at.replace(tzinfo=timezone.utc) < now:
+        raise HTTPException(status_code=410, detail="Verification code expired")
+
+    if verif.attempts >= VERIFICATION_MAX_ATTEMPTS:
+        raise HTTPException(status_code=429, detail="Too many attempts")
+
+    verif.attempts += 1
+
+    if verif.code != body.code.strip():
+        await db.commit()
+        raise HTTPException(status_code=400, detail="Invalid code")
+
+    verif.verified_at = now
+    user = await db.get(User, verif.user_id)
+    if user and user.status == "pending":
+        user.status = "active"
+    await db.commit()
+    return {"status": "verified"}
 
 
 @router.post("/login")
@@ -203,6 +293,8 @@ async def login(
             user.pw_salt = new_salt
             user.pw_version = 2
 
+    if user.status == "pending":
+        raise HTTPException(status_code=403, detail="email_verification_required")
     if user.status != "active":
         raise HTTPException(status_code=403, detail=f"Account {user.status}")
 
@@ -365,6 +457,8 @@ async def device_auth(
     user = result.scalar_one_or_none()
     if not user:
         raise HTTPException(status_code=401, detail="Invalid credentials")
+    if user.status == "pending":
+        raise HTTPException(status_code=403, detail="email_verification_required")
     if user.status != "active":
         raise HTTPException(status_code=403, detail=f"Account {user.status}")
 
@@ -514,21 +608,112 @@ async def update_profile(
     current_user: User = Depends(require_user_scope),
     db: AsyncSession = Depends(get_db),
 ):
+    pending_email = None
     if body.email is not None:
-        current_user.email = encrypt_email(body.email)
+        new_email = body.email.strip()
+        eh = hash_email_blind(new_email)
+
+        # Check that no other active/pending account uses this email
+        dup = await db.execute(
+            select(User).where(
+                User.email_hash == eh,
+                User.id != current_user.id,
+                User.status.in_(["active", "pending"]),
+            ))
+        if dup.scalar_one_or_none():
+            raise HTTPException(status_code=409, detail="Email already in use")
+
+        # Invalidate previous pending email_change verifications for this user
+        prev = await db.execute(
+            select(EmailVerification).where(
+                EmailVerification.user_id == current_user.id,
+                EmailVerification.purpose == "email_change",
+                EmailVerification.verified_at.is_(None),
+            ))
+        for old in prev.scalars().all():
+            await db.delete(old)
+
+        code = _generate_code()
+        db.add(EmailVerification(
+            email_hash=eh,
+            email_encrypted=encrypt_email(new_email),
+            code=code,
+            purpose="email_change",
+            user_id=current_user.id,
+            expires_at=datetime.now(timezone.utc) + timedelta(seconds=VERIFICATION_TTL),
+        ))
+        await db.flush()
+        mail.send_email_change_code(new_email, code)
+        pending_email = new_email
+
     await db.commit()
     email = ""
     try:
         email = decrypt_email(current_user.email) if current_user.email else ""
     except Exception:
         pass
-    return {
+    resp = {
         "user_id":  current_user.id,
         "username": current_user.username,
         "email":    email,
         "role":     current_user.role,
         "status":   current_user.status,
     }
+    if pending_email:
+        resp["email_verification_required"] = True
+    return resp
+
+
+class VerifyEmailChangeRequest(BaseModel):
+    code: str
+
+
+@router.post("/verify-email-change")
+@limiter.limit("10/minute")
+async def verify_email_change(
+    body: VerifyEmailChangeRequest,
+    request: Request,
+    current_user: User = Depends(require_user_scope),
+    db: AsyncSession = Depends(get_db),
+):
+    """Confirm an email change with the code sent to the new address."""
+    now = datetime.now(timezone.utc)
+
+    result = await db.execute(
+        select(EmailVerification).where(
+            EmailVerification.user_id == current_user.id,
+            EmailVerification.purpose == "email_change",
+            EmailVerification.verified_at.is_(None),
+        ).order_by(EmailVerification.created_at.desc()))
+    verif = result.scalar_one_or_none()
+
+    if not verif:
+        raise HTTPException(status_code=404,
+                            detail="No pending email change")
+
+    if verif.expires_at.replace(tzinfo=timezone.utc) < now:
+        raise HTTPException(status_code=410, detail="Verification code expired")
+
+    if verif.attempts >= VERIFICATION_MAX_ATTEMPTS:
+        raise HTTPException(status_code=429, detail="Too many attempts")
+
+    verif.attempts += 1
+
+    if verif.code != body.code.strip():
+        await db.commit()
+        raise HTTPException(status_code=400, detail="Invalid code")
+
+    verif.verified_at = now
+    current_user.email = verif.email_encrypted
+    current_user.email_hash = verif.email_hash
+    await db.commit()
+
+    email = ""
+    try:
+        email = decrypt_email(current_user.email) if current_user.email else ""
+    except Exception:
+        pass
+    return {"status": "verified", "email": email}
 
 
 # ── User preferences ────────────────────────────────────────────────────────
@@ -677,6 +862,7 @@ async def erase_account(db: AsyncSession, user: User) -> dict:
     await db.execute(delete(Notification).where(Notification.user_id == user.id))
     await db.execute(delete(RefreshToken).where(RefreshToken.user_id == user.id))
     await db.execute(delete(Node).where(Node.user_id == user.id))
+    await db.execute(delete(EmailVerification).where(EmailVerification.user_id == user.id))
 
     username = user.username
     # Before the name is released: the connection log is kept for its legal
@@ -686,6 +872,7 @@ async def erase_account(db: AsyncSession, user: User) -> dict:
         update(IPLog).where(IPLog.user_id == user.id).values(username=username))
     user.username = f"deleted-{user.id[:8]}"
     user.email = ""
+    user.email_hash = None
     user.pw_hash = b""
     user.pw_salt = b""
     user.pk_node_ed25519 = None