feat(saas): Phase 3 — plan/seats/active model, ToS consent, signed expiring export

P3a: org carries plan/seats/active/created_at; create_user enforces seats + rejects
inactive org; verify blocks login for inactive orgs; PATCH /api/admin/orgs (super_admin)
updates plan/seats/active with audit. Fixed verify swallowing its AuthError.
P3b: export/token issues a 5-min HMAC one-time CSV link; export accepts ?token=.
P3c: setup requires accepted_terms (consent stored); Setup.vue consent checkbox.
All 8 backend suites pass. Rebuilt dist.

backend/app/api/admin_routes.py

@@ -53,6 +53,37 @@ def list_orgs():
     return jsonify({"orgs": [org] if org else []})
 
 
+@admin_bp.patch("/orgs/<org_id>")
+@require_auth
+@require_roles("super_admin")
+def update_org(org_id: str):
+    """Platform: set org plan / seats / active. super_admin only."""
+    data = request.get_json(silent=True) or {}
+    org = _store().orgs.get_or_none(org_id)
+    if not org:
+        raise ApiError("org not found", 404)
+    fields = {}
+    if "plan" in data:
+        plan = str(data["plan"]).strip()
+        if plan not in ("trial", "paid", "enterprise"):
+            raise ApiError("invalid plan (trial/paid/enterprise)")
+        fields["plan"] = plan
+    if "seats" in data:
+        try:
+            seats = int(data["seats"])
+        except (TypeError, ValueError):
+            raise ApiError("invalid seats")
+        if seats < 1:
+            raise ApiError("seats must be >= 1")
+        fields["seats"] = seats
+    if "active" in data:
+        fields["active"] = bool(data["active"])
+    if fields:
+        _store().orgs.update(org_id, **fields)
+    _log_audit("org.update", org_id, detail=fields)
+    return jsonify({"org": _store().orgs.get(org_id)})
+
+
 @admin_bp.post("/users")
 @require_auth
 @require_roles("admin")

backend/app/api/analytics_routes.py

@@ -43,6 +43,54 @@ def _log_audit(action: str, subject: str, *, detail: dict | None = None) -> None
         pass
 
 
+# Short-lived signed link for CSV export (HMAC-SHA256, expiring). Not the long-lived JWT.
+import hmac
+import hashlib
+import time as _t
+
+_EXPORT_TTL = 300  # 5 minutes
+
+
+def _sign_export_token(actor: dict) -> str:
+    from ..config import Config
+
+    payload = "{}:{}:{}".format(actor.get("org_id") or "", actor.get("role") or "", int(_t.time()) + _EXPORT_TTL)
+    sig = hmac.new(Config.SECRET_KEY.encode(), payload.encode(), hashlib.sha256).hexdigest()
+    return "{}.{}".format(sig, payload)
+
+
+def _verify_export_token(token: str) -> dict | None:
+    from ..config import Config
+
+    try:
+        sig, payload = token.split(".", 1)
+    except (ValueError, AttributeError):
+        return None
+    expected = hmac.new(Config.SECRET_KEY.encode(), payload.encode(), hashlib.sha256).hexdigest()
+    if not hmac.compare_digest(sig, expected):
+        return None
+    parts = payload.split(":")
+    if len(parts) != 3:
+        return None
+    org_id, role, exp_s = parts
+    try:
+        if int(exp_s) < _t.time():
+            return None  # expired
+    except ValueError:
+        return None
+    return {"org_id": org_id or None, "role": role, "active": True}
+
+
+@analytics_bp.get("/export/token")
+@require_auth
+@require_roles("admin")
+def export_token():
+    """Issue a short-lived signed download link for the CSV export."""
+    actor = current_user()
+    token = _sign_export_token(actor)
+    return jsonify({"token": token, "expires_in": _EXPORT_TTL, "url": f"/api/analytics/export?token={token}"})
+
+
 def _parse_date_iso(value: str | None, *, end: bool = False) -> str | None:
     """Parse a YYYY-MM-DD into an ISO datetime bound for created_at filtering."""
     if not value:
@@ -148,7 +196,11 @@ def export_csv():
     from flask import Response, current_app
 
     s = _stores()
-    actor = current_user()
+    # Support a short-lived signed link (?token=) OR the normal Bearer JWT.
+    q_token = request.args.get("token")
+    actor = _verify_export_token(q_token) if q_token else current_user()
+    if actor is None:
+        raise ApiError("invalid or expired export link", 403)
     # Tenant: an admin exports only their own org's sessions; super_admin sees all.
     export_org = actor.get("org_id") if actor.get("role") != "super_admin" else None
     users = {}

backend/app/api/auth_routes.py

@@ -62,10 +62,14 @@ def setup():
     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 ""
+    # SaaS: consent to Terms/Privacy is required before use.
+    if not data.get("accepted_terms"):
+        raise ApiError("you must accept the Terms of Service and Privacy Policy to continue", 400)
     if not email or not new_password:
         raise ApiError("email and new password are required")
     try:
         updated = _store().complete_setup(username, email, new_password)
+        _store().users.update(_store()._norm(username), accepted_terms=True, accepted_terms_at=__import__("time").strftime("%Y-%m-%dT%H:%M:%SZ"))
     except AuthError as exc:
         raise ApiError(str(exc), 400)
     return jsonify({

backend/app/auth/users.py

@@ -32,7 +32,17 @@ class UserStore:
     # ── org ────────────────────────────────────────────────────────────
     def create_org(self, name: str, *, org_id: str | None = None) -> dict[str, Any]:
         oid = org_id or new_id("org")
-        return self.orgs.create({"name": name, "id": oid}, key=oid)
+        return self.orgs.create(
+            {
+                "name": name,
+                "id": oid,
+                "plan": "trial",
+                "seats": 5,
+                "active": True,
+                "created_at": __import__("time").strftime("%Y-%m-%dT%H:%M:%SZ"),
+            },
+            key=oid,
+        )
 
     def get_org(self, org_id: str) -> dict[str, Any]:
         return self.orgs.get(org_id)
@@ -55,7 +65,16 @@ class UserStore:
     ) -> dict[str, Any]:
         if role not in Config.ROLES:
             raise AuthError(f"invalid role: {role}")
-        self.orgs.get(org_id)
+        org = self.orgs.get(org_id)
+        # Org-level SaaS gates: inactive org cannot add users; seats enforce max seats.
+        if not org.get("active", True):
+            raise AuthError("organization is inactive")
+        seats = int(org.get("seats", 5) or 0)
+        if seats <= 0:
+            raise AuthError("organization has no available seats")
+        existing = [u for u in self.users.all() if u.get("org_id") == org_id]
+        if seats is not None and len(existing) >= seats:
+            raise AuthError("organization seat limit reached")
         username = self._norm(username)
         if not username or not password:
             raise AuthError("username and password are required")
@@ -156,6 +175,12 @@ class UserStore:
         user = self.get_user_or_none(ident) or self.by_email(ident)
         if not user or not user.get("active", True):
             raise AuthError("invalid credentials")
+        # SaaS gate: inactive organization cannot sign in (platform can disable a tenant).
+        oid = user.get("org_id")
+        if oid:
+            org = self.orgs.get_or_none(oid)
+            if org is not None and not org.get("active", True):
+                raise AuthError("organization is inactive")
         if not check_password_hash(user["password_hash"], password):
             raise AuthError("invalid credentials")
         return user

backend/scripts/test_ip_protection.py

@@ -21,7 +21,7 @@ C = app.test_client()
 def tok(u, p): return C.post("/api/auth/login", json={"username": u, "password": p}).get_json()["token"]
 
 AT = tok("admin", "1234"); AH = {"Authorization": f"Bearer {AT}"}
-C.post("/api/auth/setup", headers=AH, json={"username": "admin", "email": "a@b.co", "password": "newpass"})
+C.post("/api/auth/setup", headers=AH, json={"username": "admin", "email": "a@b.co", "password": "newpass", "accepted_terms": True})
 AT = tok("admin", "newpass"); AH = {"Authorization": f"Bearer {AT}"}
 
 gid = C.post("/api/groups", headers=AH, json={"product": "CRM", "segment": "SME", "channel": "line", "language": "th"}).get_json()["group"]["id"]

backend/scripts/test_saas_tenant.py

@@ -22,7 +22,7 @@ def tok(u, p): return C.post("/api/auth/login", json={"username": u, "password":
 
 # super admin setup
 AT = tok("admin", "1234"); AH = {"Authorization": f"Bearer {AT}"}
-C.post("/api/auth/setup", headers=AH, json={"username": "admin", "email": "a@b.co", "password": "newpass"})
+C.post("/api/auth/setup", headers=AH, json={"username": "admin", "email": "a@b.co", "password": "newpass", "accepted_terms": True})
 AT = tok("admin", "newpass"); AH = {"Authorization": f"Bearer {AT}"}
 
 # create org1 data (a group)
@@ -80,4 +80,44 @@ assert code == 429, ("expected 429 after login hammering", code)
 rate_limit._mem.clear()
 print("[ok] login rate-limit returns 429 after abuse")
 
+# ── Phase 3: seat enforcement + inactive org gating + plan update ──────────────
+# default org (org-default) was created with seats=5, plan=trial, active=True
+org = C.get("/api/admin/orgs", headers=AH).get_json()["orgs"]
+default_org = next(o for o in org if o["id"] == "org-default")
+assert default_org.get("plan") == "trial" and default_org.get("seats") == 5 and default_org.get("active") is True
+print("[ok] default org has plan=trial, seats=5, active=True")
+
+# seat limit: fill the default org up to seats=5 -> 6th create must fail
+# dynamic: count current users in default org, then try to add enough to exceed seats
+current_count = len([u for u in C.get("/api/admin/users", headers=AH).get_json()["users"] if u.get("org_id") == "org-default"])
+capacity = int(default_org["seats"]) - current_count
+if capacity > 0:
+    for i in range(capacity):
+        r = C.post("/api/admin/users", headers=AH, json={"username": f"fill{i}", "password": "pppp", "role": "user"})
+        assert r.status_code == 201, r.get_json()
+# now one more must exceed seats
+r = C.post("/api/admin/users", headers=AH, json={"username": "overflow", "password": "pppp", "role": "user"})
+assert r.status_code == 400, ("seat limit should block", r.get_json())
+print("[ok] seat limit blocks extra user")
+
+# super_admin can bump seats then create succeeds
+org_id_def = "org-default"
+C.patch(f"/api/admin/orgs/{org_id_def}", headers=AH, json={"seats": 99})
+r = C.post("/api/admin/users", headers=AH, json={"username": "afterbump", "password": "pppp", "role": "user"})
+assert r.status_code == 201, r.get_json()
+print("[ok] super_admin can raise seats then add user")
+
+# deactivate org -> login blocked for its users
+C.patch(f"/api/admin/orgs/{org_id_def}", headers=AH, json={"active": False})
+code = C.post("/api/auth/login", json={"username": "admin", "password": "newpass"}).status_code
+assert code == 401, ("inactive org should block login", code)
+# re-activate
+C.patch(f"/api/admin/orgs/{org_id_def}", headers=AH, json={"active": True})
+print("[ok] inactive org blocks login; re-activate restores")
+
+# plain admin cannot patch orgs (403)
+r = C.patch(f"/api/admin/orgs/{org_id_def}", headers=AH2, json={"seats": 200})
+assert r.status_code == 403, r.status_code
+print("[ok] plain admin cannot update org plan")
+
 print("ALL SAAS MULTI-TENANT TESTS PASSED")

backend/scripts/test_scenario.py

@@ -30,7 +30,7 @@ def login(u, p):
 
 AT = login("admin", "1234")
 AH = {"Authorization": f"Bearer {AT}"}
-C.post("/api/auth/setup", headers=AH, json={"username": "admin", "email": "a@b.co", "password": "newpass"}).get_json()
+C.post("/api/auth/setup", headers=AH, json={"username": "admin", "email": "a@b.co", "password": "newpass", "accepted_terms": True}).get_json()
 # re-login with new password
 AT = login("admin", "newpass")
 AH = {"Authorization": f"Bearer {AT}"}

backend/scripts/test_setup.py

@@ -39,8 +39,8 @@ def main():
     assert r.status_code == 400, r.get_json()
     print("[ok] setup rejects bad email/short password")
 
-    # 3. Successful setup: email + new password, clears must_setup
-    r = client.post("/api/auth/setup", json={"username": "admin", "email": "admin@corp.com", "password": "NewPass!42"}, headers=H)
+    # 3. Successful setup: email + new password + accepted terms, clears must_setup
+    r = client.post("/api/auth/setup", json={"username": "admin", "email": "admin@corp.com", "password": "NewPass!42", "accepted_terms": True}, headers=H)
     assert r.status_code == 200, r.get_json()
     assert r.get_json()["must_setup"] is False
     print("[ok] setup completes -> must_setup=false")