c3d31c06e2
- Auth/roles (no self-reg), admin user provision, JWT - Analyze: sales kit + initial pain-fit from form/upload - Persona generator: 15 personas (5/tier) w/ pain variety, negotiation, init mode, channel, latent/revealable, wrong_text special - Chat simulator: per-mode initiation, one-shot, hidden signals, judge-LLM debrief+coaching - Trainee loop: win/lose board, weak-areas, user-generated personas - Admin analytics; EN+TH Vue SPA served by Flask - Deploy: Dockerfile, docker-compose, README, eng-log + HANDOFF - Tests (mock LLM): m0/m1/routes/e2e all pass
129 lines
4.9 KiB
Python
129 lines
4.9 KiB
Python
"""User + organization store and auth logic (JWT, password hashing, roles)."""
|
|
from __future__ import annotations
|
|
|
|
import datetime
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import jwt
|
|
from werkzeug.security import check_password_hash, generate_password_hash
|
|
|
|
from ..config import Config
|
|
from ..storage.store import JsonStore, StoreError, new_id
|
|
|
|
|
|
class AuthError(Exception):
|
|
pass
|
|
|
|
|
|
class UserStore:
|
|
def __init__(self, data_dir: Path) -> None:
|
|
self.users = JsonStore(data_dir / "users")
|
|
self.orgs = JsonStore(data_dir / "orgs")
|
|
|
|
# ── org ────────────────────────────────────────────────────────────
|
|
def create_org(self, name: str, *, org_id: str | None = None) -> dict[str, Any]:
|
|
return self.orgs.create(
|
|
{"name": name, "id": org_id or new_id("org")},
|
|
key=org_id or new_id("org"),
|
|
)
|
|
|
|
def get_org(self, org_id: str) -> dict[str, Any]:
|
|
return self.orgs.get(org_id)
|
|
|
|
# ── users ──────────────────────────────────────────────────────────
|
|
def create_user(
|
|
self,
|
|
*,
|
|
org_id: str,
|
|
email: str,
|
|
password: str,
|
|
name: str,
|
|
role: str = "user",
|
|
) -> 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")
|
|
user = {
|
|
"id": email, # email = unique id/username
|
|
"email": email,
|
|
"org_id": org_id,
|
|
"org_name": org.get("name", ""),
|
|
"name": name.strip() or email,
|
|
"password_hash": generate_password_hash(password),
|
|
"role": role,
|
|
"created_at": datetime.datetime.now(datetime.timezone.utc).isoformat(),
|
|
"active": True,
|
|
}
|
|
return self.users.create(user, key=email)
|
|
|
|
def get_user(self, email: str) -> dict[str, Any]:
|
|
email = email.strip().lower()
|
|
return self.users.get(email)
|
|
|
|
def get_user_or_none(self, email: str) -> dict[str, Any] | None:
|
|
return self.users.get_or_none(email.strip().lower())
|
|
|
|
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_role(self, email: 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)
|
|
|
|
def set_password(self, email: str, new_password: str) -> dict[str, Any]:
|
|
if not new_password:
|
|
raise AuthError("password is required")
|
|
return self.users.update(
|
|
email.strip().lower(),
|
|
password_hash=generate_password_hash(new_password),
|
|
)
|
|
|
|
# ── auth ───────────────────────────────────────────────────────────
|
|
def verify(self, email: str, password: str) -> dict[str, Any]:
|
|
user = self.get_user_or_none(email)
|
|
if not user or not user.get("active", True):
|
|
raise AuthError("invalid credentials")
|
|
if not check_password_hash(user["password_hash"], password):
|
|
raise AuthError("invalid credentials")
|
|
return user
|
|
|
|
def issue_token(self, user: dict[str, Any]) -> str:
|
|
now = datetime.datetime.now(datetime.timezone.utc)
|
|
payload = {
|
|
"sub": user["email"],
|
|
"org_id": user["org_id"],
|
|
"role": user["role"],
|
|
"iat": now,
|
|
"exp": now + datetime.timedelta(hours=Config.JWT_EXPIRES_HOURS),
|
|
}
|
|
return jwt.encode(payload, Config.SECRET_KEY, algorithm=Config.JWT_ALGO)
|
|
|
|
def decode_token(self, token: str) -> dict[str, Any]:
|
|
try:
|
|
return jwt.decode(
|
|
token, Config.SECRET_KEY, algorithms=[Config.JWT_ALGO]
|
|
)
|
|
except jwt.PyJWTError as exc:
|
|
raise AuthError("invalid or expired token") from exc
|
|
|
|
def public_user(self, user: dict[str, Any]) -> dict[str, Any]:
|
|
u = dict(user)
|
|
u.pop("password_hash", None)
|
|
return u
|