Add username-based login + mandatory first-time admin setup; push-ready
- User id/login = username (was email). Email is a separate settable field. - Default admin: username admin / password 1234, must_setup=True. - Login forces /setup on first login: set email + change password, then clears must_setup. - New /api/auth/setup endpoint; JWT sub = username; admin routes use username. - Frontend: Login uses username, router guard forces /setup, new Setup.vue (email + new password + confirm), i18n EN/TH. - Tests: test_setup.py added; all suites adapted (m0/m1/routes/security/setup/e2e) PASS.
This commit is contained in:
@@ -23,13 +23,13 @@ def create_user():
|
||||
"""Create a user + provision a password (invite). Admin or super-admin only."""
|
||||
data = request.get_json(silent=True) or {}
|
||||
name = (data.get("name") or "").strip()
|
||||
email = (data.get("email") or "").strip().lower()
|
||||
username = (data.get("username") or data.get("email") or "").strip().lower()
|
||||
password = data.get("password") or ""
|
||||
role = (data.get("role") or "user").strip()
|
||||
org_id = (data.get("org_id") or current_user().get("org_id") or "org-default").strip()
|
||||
|
||||
if not email or not password:
|
||||
raise ApiError("email and password are required")
|
||||
if not username or not password:
|
||||
raise ApiError("username and password are required")
|
||||
if role not in Config.ROLES:
|
||||
raise ApiError(f"invalid role: {role}")
|
||||
# Only super_admin can create another admin/super_admin
|
||||
@@ -38,7 +38,7 @@ def create_user():
|
||||
raise ApiError("only super_admin can grant admin roles", 403)
|
||||
try:
|
||||
user = _store().create_user(
|
||||
org_id=org_id, email=email, password=password, name=name, role=role
|
||||
org_id=org_id, username=username, password=password, name=name, role=role
|
||||
)
|
||||
except AuthError as exc:
|
||||
raise ApiError(str(exc))
|
||||
@@ -57,14 +57,14 @@ def list_users():
|
||||
return jsonify({"users": users})
|
||||
|
||||
|
||||
@admin_bp.put("/users/<email>")
|
||||
@admin_bp.put("/users/<username>")
|
||||
@require_auth
|
||||
@require_roles("admin")
|
||||
def update_user(email: str):
|
||||
def update_user(username: str):
|
||||
data = request.get_json(silent=True) or {}
|
||||
email = email.strip().lower()
|
||||
username = username.strip().lower()
|
||||
actor = current_user()
|
||||
target = _store().get_user_or_none(email)
|
||||
target = _store().get_user_or_none(username)
|
||||
if not target:
|
||||
raise ApiError("user not found", 404)
|
||||
|
||||
@@ -75,14 +75,20 @@ def update_user(email: str):
|
||||
raise ApiError(f"invalid role: {role}")
|
||||
if actor.get("role") != "super_admin":
|
||||
raise ApiError("only super_admin can change roles")
|
||||
_store().set_role(email, role)
|
||||
_store().set_role(username, role)
|
||||
|
||||
if "active" in data:
|
||||
if actor.get("role") != "super_admin":
|
||||
raise ApiError("only super_admin can activate/deactivate users")
|
||||
_store().set_active(email, bool(data.get("active")))
|
||||
_store().set_active(username, bool(data.get("active")))
|
||||
|
||||
if "password" in data and data.get("password"):
|
||||
_store().set_password(email, data.get("password"))
|
||||
_store().set_password(username, data.get("password"))
|
||||
|
||||
return jsonify({"user": _store().public_user(_store().get_user(email))})
|
||||
if "email" in data:
|
||||
try:
|
||||
_store().set_email(username, data.get("email"))
|
||||
except AuthError as exc:
|
||||
raise ApiError(str(exc))
|
||||
|
||||
return jsonify({"user": _store().public_user(_store().get_user(username))})
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
"""Auth routes: login + current user. No self-registration."""
|
||||
"""Auth routes: login, current user, first-time admin setup. No self-registration."""
|
||||
from __future__ import annotations
|
||||
|
||||
from flask import Blueprint, jsonify, request
|
||||
@@ -15,22 +15,53 @@ def _store():
|
||||
return current_app.extensions["user_store"]
|
||||
|
||||
|
||||
def _login_body(data: dict) -> str:
|
||||
# Accept `username` (primary) or `email` (fallback), lower-cased.
|
||||
return (data.get("username") or data.get("email") or "").strip().lower()
|
||||
|
||||
|
||||
@auth_bp.post("/login")
|
||||
def login():
|
||||
data = request.get_json(silent=True) or {}
|
||||
email = (data.get("email") or "").strip().lower()
|
||||
username = _login_body(data)
|
||||
password = data.get("password") or ""
|
||||
if not email or not password:
|
||||
raise ApiError("email and password are required")
|
||||
if not username or not password:
|
||||
raise ApiError("username and password are required")
|
||||
try:
|
||||
user = _store().verify(email, password)
|
||||
user = _store().verify(username, password)
|
||||
token = _store().issue_token(user)
|
||||
except AuthError as exc:
|
||||
raise ApiError(str(exc), 401)
|
||||
return jsonify({"token": token, "user": _store().public_user(user)})
|
||||
return jsonify({
|
||||
"token": token,
|
||||
"user": _store().public_user(user),
|
||||
"must_setup": bool(user.get("must_setup")),
|
||||
})
|
||||
|
||||
|
||||
@auth_bp.get("/me")
|
||||
@require_auth
|
||||
def me():
|
||||
return jsonify({"user": _store().public_user(current_user())})
|
||||
|
||||
|
||||
@auth_bp.post("/setup")
|
||||
@require_auth
|
||||
def setup():
|
||||
"""First-time admin setup: set email + change password, then clear must_setup."""
|
||||
user = current_user()
|
||||
data = request.get_json(silent=True) or {}
|
||||
username = (data.get("username") or user.get("username") or user.get("id") or "").strip().lower()
|
||||
email = (data.get("email") or "").strip()
|
||||
new_password = data.get("password") or ""
|
||||
if not email or not new_password:
|
||||
raise ApiError("email and new password are required")
|
||||
try:
|
||||
updated = _store().complete_setup(username, email, new_password)
|
||||
except AuthError as exc:
|
||||
raise ApiError(str(exc), 400)
|
||||
return jsonify({
|
||||
"ok": True,
|
||||
"user": _store().public_user(updated),
|
||||
"must_setup": False,
|
||||
})
|
||||
|
||||
@@ -1,7 +1,13 @@
|
||||
"""User + organization store and auth logic (JWT, password hashing, roles)."""
|
||||
"""User + organization store and auth logic (JWT, password hashing, roles).
|
||||
|
||||
Login identity is the user's `username` (stable id). `email` is an optional
|
||||
separate field that admins/users can set; the default admin must set an email
|
||||
+before first real use (enforced via `must_setup`).
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import datetime
|
||||
import re
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
@@ -11,6 +17,8 @@ from werkzeug.security import check_password_hash, generate_password_hash
|
||||
from ..config import Config
|
||||
from ..storage.store import JsonStore, StoreError, new_id
|
||||
|
||||
EMAIL_RE = re.compile(r"^[^@\s]+@[^@\s]+\.[^@\s]+$")
|
||||
|
||||
|
||||
class AuthError(Exception):
|
||||
pass
|
||||
@@ -32,71 +40,114 @@ class UserStore:
|
||||
return self.orgs.get(org_id)
|
||||
|
||||
# ── users ──────────────────────────────────────────────────────────
|
||||
@staticmethod
|
||||
def _norm(username: str) -> str:
|
||||
return username.strip().lower()
|
||||
|
||||
def create_user(
|
||||
self,
|
||||
*,
|
||||
org_id: str,
|
||||
email: str,
|
||||
username: str,
|
||||
password: str,
|
||||
name: str,
|
||||
role: str = "user",
|
||||
email: str | None = None,
|
||||
must_setup: bool = False,
|
||||
) -> dict[str, Any]:
|
||||
if role not in Config.ROLES:
|
||||
raise AuthError(f"invalid role: {role}")
|
||||
org = self.orgs.get(org_id)
|
||||
email = email.strip().lower()
|
||||
if not email or not password:
|
||||
raise AuthError("email and password are required")
|
||||
if self.users.get_or_none(email) is not None:
|
||||
raise AuthError("a user with this email already exists")
|
||||
self.orgs.get(org_id)
|
||||
username = self._norm(username)
|
||||
if not username or not password:
|
||||
raise AuthError("username and password are required")
|
||||
if not re.fullmatch(r"[a-zA-Z0-9_.-]{2,64}", username):
|
||||
raise AuthError("invalid username (letters/numbers/._- only, 2-64 chars)")
|
||||
if self.users.get_or_none(username) is not None:
|
||||
raise AuthError("a user with this username already exists")
|
||||
email = (email or "").strip().lower() or None
|
||||
if email:
|
||||
if not EMAIL_RE.fullmatch(email):
|
||||
raise AuthError("invalid email")
|
||||
if self.email_exists(email):
|
||||
raise AuthError("a user with this email already exists")
|
||||
user = {
|
||||
"id": email, # email = unique id/username
|
||||
"id": username,
|
||||
"username": username,
|
||||
"email": email,
|
||||
"org_id": org_id,
|
||||
"org_name": org.get("name", ""),
|
||||
"name": name.strip() or email,
|
||||
"org_name": self.orgs.get(org_id).get("name", ""),
|
||||
"name": name.strip() or username,
|
||||
"password_hash": generate_password_hash(password),
|
||||
"role": role,
|
||||
"must_setup": must_setup,
|
||||
"created_at": datetime.datetime.now(datetime.timezone.utc).isoformat(),
|
||||
"active": True,
|
||||
}
|
||||
return self.users.create(user, key=email)
|
||||
return self.users.create(user, key=username)
|
||||
|
||||
def get_user(self, email: str) -> dict[str, Any]:
|
||||
email = email.strip().lower()
|
||||
return self.users.get(email)
|
||||
def get_user(self, username: str) -> dict[str, Any]:
|
||||
return self.users.get(self._norm(username))
|
||||
|
||||
def get_user_or_none(self, email: str) -> dict[str, Any] | None:
|
||||
return self.users.get_or_none(email.strip().lower())
|
||||
def get_user_or_none(self, username: str) -> dict[str, Any] | None:
|
||||
return self.users.get_or_none(self._norm(username))
|
||||
|
||||
def by_email(self, email: str) -> dict[str, Any] | None:
|
||||
email = (email or "").strip().lower()
|
||||
if not email:
|
||||
return None
|
||||
for u in self.users.all():
|
||||
if u.get("email") and u["email"] == email:
|
||||
return u
|
||||
return None
|
||||
|
||||
def email_exists(self, email: str) -> bool:
|
||||
return self.by_email(email) is not None
|
||||
|
||||
def list_users(self, *, org_id: str | None = None) -> list[dict[str, Any]]:
|
||||
users = self.users.all()
|
||||
if org_id:
|
||||
users = [u for u in users if u.get("org_id") == org_id]
|
||||
# Redact password hash
|
||||
for u in users:
|
||||
u.pop("password_hash", None)
|
||||
return users
|
||||
|
||||
def set_active(self, email: str, active: bool) -> dict[str, Any]:
|
||||
return self.users.update(email.strip().lower(), active=active)
|
||||
def set_active(self, username: str, active: bool) -> dict[str, Any]:
|
||||
return self.users.update(self._norm(username), active=active)
|
||||
|
||||
def set_role(self, email: str, role: str) -> dict[str, Any]:
|
||||
def set_role(self, username: str, role: str) -> dict[str, Any]:
|
||||
if role not in Config.ROLES:
|
||||
raise AuthError(f"invalid role: {role}")
|
||||
return self.users.update(email.strip().lower(), role=role)
|
||||
return self.users.update(self._norm(username), role=role)
|
||||
|
||||
def set_password(self, email: str, new_password: str) -> dict[str, Any]:
|
||||
def set_password(self, username: str, new_password: str) -> dict[str, Any]:
|
||||
if not new_password:
|
||||
raise AuthError("password is required")
|
||||
return self.users.update(
|
||||
email.strip().lower(),
|
||||
self._norm(username),
|
||||
password_hash=generate_password_hash(new_password),
|
||||
)
|
||||
|
||||
def set_email(self, username: str, email: str) -> dict[str, Any]:
|
||||
email = (email or "").strip().lower()
|
||||
if not EMAIL_RE.fullmatch(email):
|
||||
raise AuthError("invalid email")
|
||||
existing = self.by_email(email)
|
||||
if existing and existing["id"] != self._norm(username):
|
||||
raise AuthError("a user with this email already exists")
|
||||
return self.users.update(self._norm(username), email=email)
|
||||
|
||||
def complete_setup(self, username: str, email: str, new_password: str) -> dict[str, Any]:
|
||||
"""First-time admin setup: set email + password, clear must_setup."""
|
||||
if not new_password or len(new_password) < 4:
|
||||
raise AuthError("password must be at least 4 characters")
|
||||
self.set_email(username, email)
|
||||
self.set_password(username, new_password)
|
||||
return self.users.update(self._norm(username), must_setup=False)
|
||||
|
||||
# ── auth ───────────────────────────────────────────────────────────
|
||||
def verify(self, email: str, password: str) -> dict[str, Any]:
|
||||
user = self.get_user_or_none(email)
|
||||
def verify(self, username: str, password: str) -> dict[str, Any]:
|
||||
user = self.get_user_or_none(username)
|
||||
if not user or not user.get("active", True):
|
||||
raise AuthError("invalid credentials")
|
||||
if not check_password_hash(user["password_hash"], password):
|
||||
@@ -106,7 +157,7 @@ class UserStore:
|
||||
def issue_token(self, user: dict[str, Any]) -> str:
|
||||
now = datetime.datetime.now(datetime.timezone.utc)
|
||||
payload = {
|
||||
"sub": user["email"],
|
||||
"sub": user.get("username") or user.get("id"),
|
||||
"org_id": user["org_id"],
|
||||
"role": user["role"],
|
||||
"iat": now,
|
||||
@@ -116,9 +167,7 @@ class UserStore:
|
||||
|
||||
def decode_token(self, token: str) -> dict[str, Any]:
|
||||
try:
|
||||
return jwt.decode(
|
||||
token, Config.SECRET_KEY, algorithms=[Config.JWT_ALGO]
|
||||
)
|
||||
return jwt.decode(token, Config.SECRET_KEY, algorithms=[Config.JWT_ALGO])
|
||||
except jwt.PyJWTError as exc:
|
||||
raise AuthError("invalid or expired token") from exc
|
||||
|
||||
|
||||
@@ -11,20 +11,24 @@ from .config import Config
|
||||
|
||||
|
||||
def bootstrap_admin(users: UserStore) -> None:
|
||||
"""Ensure a default org + super-admin exists on first run (no self-registration)."""
|
||||
email = "admin@salestrainer.local"
|
||||
"""Ensure a default org + super-admin exists on first run (no self-registration).
|
||||
|
||||
Default admin logs in with username `admin` / `1234`, then MUST set an email
|
||||
and change the password on first login (`must_setup=True`).
|
||||
"""
|
||||
org = users.orgs.get_or_none("org-default")
|
||||
if org is None:
|
||||
org = users.create_org("Default Organization", org_id="org-default")
|
||||
if users.get_user_or_none(email) is None:
|
||||
if users.get_user_or_none("admin") is None:
|
||||
users.create_user(
|
||||
org_id=org["id"],
|
||||
email=email,
|
||||
password="admin123",
|
||||
username="admin",
|
||||
password="1234",
|
||||
name="Super Admin",
|
||||
role="super_admin",
|
||||
must_setup=True,
|
||||
)
|
||||
print("[bootstrap] created default super-admin:", email, "/ admin123")
|
||||
print("[bootstrap] created default super-admin: admin / 1234 (must set email + password)")
|
||||
|
||||
|
||||
def create_app() -> Flask:
|
||||
|
||||
Reference in New Issue
Block a user