Files
sales-trainer/backend/tests/test_demo_accounts.py
Macky 3c22d88bcd feat: demo SaaS + training flow security hardening (8/8 review gate passed)
- Demo accounts: super_admin-only provisioning into isolated DEMO_ORG_ID tenant,
  30-day UTC trial on first login, revocable, one-time credential delivery via
  optional SES/webhook (never persisted). Adds boto3 dependency.
- Analytics/report/export/privacy: shared bounded scan budget across users/groups/
  sessions, tenant-consistent session/user/group joins, scalar-only CSV export
  (no nested persisted-value stringification).
- Ownership/tenant isolation: canonical owner-tenant predicate for list/read/chat;
  client sees is_owned only, never owner_user_id.
- Lifecycle/races: status transition validation, analyzing is an in-progress gate
  (no duplicate reanalysis), structured-ready publication, stale-variant revalidation.
- Auth/setup/consent/JWT/OAuth/config: fail-closed consent, bounded JWT lifetime,
  provider-subject atomic OAuth identity, repeated-secret rejection, strict Persona
  trait validation.
- Chat/session/privacy: pre-seller opener redaction, corrupt-session recovery,
  role-aware completed-chat dashboard routing.
- Frontend: Training→product→personas→practice flow, demo/role/demo guards,
  is_owned-based ownership display, 320×568 and 500×768 responsive E2E.
- 8 independent exact-five-key review scopes passed; backend 509, frontend 26,
  production build 1775 modules, isolated E2E 15.
2026-08-25 06:39:06 +07:00

585 lines
22 KiB
Python

"""Focused contracts for demo accounts, visibility gates, and delivery safety."""
from __future__ import annotations
import datetime as dt
import json
from urllib.error import URLError
from urllib.request import Request
import pytest
from app.auth import users as users_module
from app.auth.users import AuthError
from app.config import Config
from app.services.demo_delivery import _HttpsOnlyRedirectHandler
from scripts.mock_llm import MockLLM
FIXED_NOW = dt.datetime(2026, 8, 23, 12, 0, tzinfo=dt.timezone.utc)
def _headers(token: str) -> dict[str, str]:
return {"Authorization": f"Bearer {token}"}
def _super_admin_token(user_store, login) -> str:
user_store.complete_setup(
"admin",
"admin@example.com",
"admin-ready-password",
accepted_terms=True,
accepted_terms_at="2026-08-23T00:00:00Z",
)
return login("admin", "admin-ready-password")["token"]
def _ready_group(app, *, org_id: str, creator_id: str, visibility: str, suffix: str) -> tuple[str, str]:
groups = app.extensions["group_store"]
group = groups.create(
org_id=org_id,
creator_id=creator_id,
title=f"{visibility}-{suffix}",
visibility=visibility,
input_data={"product": f"Product {suffix}", "channel": "facebook", "language": "en"},
sales_kit={"productName": f"Product {suffix}"},
report={"summary": "Test report"},
)
persona_id = f"persona-{suffix}"
groups.update(
group["id"],
status="ready",
personas=[
{
"id": persona_id,
"name": f"Customer {suffix}",
"tier": "A",
"channel": "facebook",
"initiation_mode": "customer",
"opener": "Hello, I have a question.",
}
],
)
return group["id"], persona_id
def _active_demo_token(user_store, *, username: str, email: str) -> tuple[dict, str]:
user_store.ensure_demo_org()
user = user_store.create_demo_user(
email=email,
username=username,
password="demo-original-password",
)
# This helper is for business-route authorization tests; the HTTP endpoint
# separately covers the required first-time setup flow.
user_store.users.update(
username,
must_setup=False,
accepted_terms=True,
accepted_terms_at=FIXED_NOW.isoformat(),
)
active = user_store.start_demo_trial(username, now=FIXED_NOW)
return active, user_store.issue_token(active)
def test_demo_creation_is_super_admin_only_and_generic_demo_creation_is_blocked(
client, user_store, login
):
super_token = _super_admin_token(user_store, login)
admin = user_store.create_user(
org_id="org-default",
username="tenant-admin-demo-gate",
password="tenant-admin-demo-password",
name="Tenant Admin",
role="admin",
)
user = user_store.create_user(
org_id="org-default",
username="plain-demo-gate-user",
password="plain-demo-gate-password",
name="Plain User",
role="user",
)
admin_token = login(admin["username"], "tenant-admin-demo-password")["token"]
user_token = login(user["username"], "plain-demo-gate-password")["token"]
for token in (admin_token, user_token):
response = client.post(
"/api/admin/demo-accounts",
json={"email": "blocked-demo@example.com"},
headers=_headers(token),
)
assert response.status_code == 403, response.get_json()
extra_field = client.post(
"/api/admin/demo-accounts",
json={"email": "strict-demo@example.com", "name": "ignored"},
headers=_headers(super_token),
)
assert extra_field.status_code == 400, extra_field.get_json()
generic = client.post(
"/api/admin/users",
json={
"username": "generic-demo-user",
"password": "generic-demo-password",
"role": "demo",
},
headers=_headers(super_token),
)
assert generic.status_code == 403, generic.get_json()
with pytest.raises(AuthError, match="dedicated provisioning"):
user_store.create_user(
org_id=Config.DEMO_ORG_ID,
username="service-generic-demo",
password="service-generic-password",
name="Service Generic Demo",
role="demo",
)
def test_demo_creation_rejects_duplicate_email(client, user_store, login):
token = _super_admin_token(user_store, login)
first = client.post(
"/api/admin/demo-accounts",
json={"email": "Duplicate-Demo@example.com"},
headers=_headers(token),
)
second = client.post(
"/api/admin/demo-accounts",
json={"email": "duplicate-demo@example.com"},
headers=_headers(token),
)
assert first.status_code == 201, first.get_json()
assert second.status_code == 409, second.get_json()
assert len(
[u for u in user_store.users.all() if u.get("email") == "duplicate-demo@example.com"]
) == 1
def test_demo_creation_delivery_is_mocked_and_plaintext_password_is_never_persisted(
client, user_store, login, tmp_path, monkeypatch, caplog
):
token = _super_admin_token(user_store, login)
delivery = client.application.extensions["demo_delivery"]
temporary_password = "one-time-demo-password-never-persisted"
monkeypatch.setattr(delivery, "generate_password", lambda: temporary_password)
monkeypatch.setattr(
delivery,
"_send_email",
lambda **kwargs: {"status": "mocked", "provider": "test", "attempts": 1},
)
monkeypatch.setattr(
delivery,
"_send_webhook",
lambda **kwargs: {"status": "mocked", "provider": "test", "attempts": 1},
)
response = client.post(
"/api/admin/demo-accounts",
json={"email": "delivery-safe@example.com"},
headers=_headers(token),
)
assert response.status_code == 201, response.get_json()
assert response.get_json()["temporary_password"] == temporary_password
body = response.get_json()
username = body["username"]
stored = user_store.get_user(username)
assert temporary_password not in stored["password_hash"]
persisted_text = []
for path in tmp_path.rglob("*"):
if path.is_file():
persisted_text.append(path.read_text(encoding="utf-8", errors="ignore"))
all_persisted = "\n".join(persisted_text)
assert temporary_password not in all_persisted
assert "temporary_password" not in all_persisted
assert temporary_password not in "\n".join(record.getMessage() for record in caplog.records)
events = client.application.extensions["demo_delivery"].events.all()
assert len(events) == 1
assert temporary_password not in json.dumps(events[0], ensure_ascii=False)
def test_demo_first_login_starts_one_exact_30_day_trial_and_is_idempotent(
client, user_store, login, monkeypatch
):
token = _super_admin_token(user_store, login)
delivery = client.application.extensions["demo_delivery"]
temporary_password = "first-login-demo-password"
monkeypatch.setattr(delivery, "generate_password", lambda: temporary_password)
monkeypatch.setattr(
delivery,
"deliver",
lambda **kwargs: {
"event_id": "test-event",
"email": {"status": "mocked", "attempts": 1},
"webhook": {"status": "mocked", "attempts": 1},
},
)
monkeypatch.setattr(users_module, "utc_now", lambda: FIXED_NOW)
created = client.post(
"/api/admin/demo-accounts",
json={"email": "trial-clock@example.com"},
headers=_headers(token),
)
assert created.status_code == 201, created.get_json()
username = created.get_json()["username"]
assert created.get_json()["user"]["trial_started_at"] is None
assert created.get_json()["user"]["trial_expires_at"] is None
first_login = client.post(
"/api/auth/login",
json={"username": username, "password": temporary_password},
)
second_login = client.post(
"/api/auth/login",
json={"username": username, "password": temporary_password},
)
assert first_login.status_code == 200, first_login.get_json()
assert second_login.status_code == 200, second_login.get_json()
persisted = user_store.get_user(username)
started = dt.datetime.fromisoformat(persisted["trial_started_at"])
expires = dt.datetime.fromisoformat(persisted["trial_expires_at"])
assert started == FIXED_NOW
assert expires - started == dt.timedelta(days=30)
assert second_login.get_json()["user"]["trial_started_at"] == persisted["trial_started_at"]
assert second_login.get_json()["user"]["trial_expires_at"] == persisted["trial_expires_at"]
def test_malformed_or_expired_demo_state_rejects_login_and_existing_tokens(
client, user_store, monkeypatch
):
monkeypatch.setattr(users_module, "utc_now", lambda: FIXED_NOW)
malformed, malformed_token = _active_demo_token(
user_store,
username="malformed-demo-state",
email="malformed-demo-state@example.com",
)
user_store.users.update(
malformed["username"],
trial_started_at="not-a-timestamp",
trial_expires_at=None,
)
malformed_me = client.get("/api/auth/me", headers=_headers(malformed_token))
malformed_login = client.post(
"/api/auth/login",
json={"username": malformed["username"], "password": "demo-original-password"},
)
assert malformed_me.status_code == 401, malformed_me.get_json()
assert malformed_login.status_code == 401, malformed_login.get_json()
expired, expired_token = _active_demo_token(
user_store,
username="expired-demo-state",
email="expired-demo-state@example.com",
)
expired_started = FIXED_NOW - dt.timedelta(days=31)
expired_expires = FIXED_NOW - dt.timedelta(days=1)
user_store.users.update(
expired["username"],
trial_started_at=expired_started.isoformat(),
trial_expires_at=expired_expires.isoformat(),
)
expired_me = client.get("/api/auth/me", headers=_headers(expired_token))
expired_login = client.post(
"/api/auth/login",
json={"username": expired["username"], "password": "demo-original-password"},
)
assert expired_me.status_code == 401, expired_me.get_json()
assert expired_login.status_code == 401, expired_login.get_json()
def test_revoke_demo_invalidates_an_existing_token(client, user_store, login):
super_token = _super_admin_token(user_store, login)
demo, demo_token = _active_demo_token(
user_store,
username="revoked-demo-account",
email="revoked-demo-account@example.com",
)
before_version = demo["auth_version"]
revoked = client.post(
f"/api/admin/demo-accounts/{demo['username']}/revoke",
headers=_headers(super_token),
)
assert revoked.status_code == 200, revoked.get_json()
updated = user_store.get_user(demo["username"])
assert updated["active"] is False
assert updated["auth_version"] == before_version + 1
assert revoked.get_json()["revoked"] is True
existing_token_response = client.get(
"/api/auth/me", headers=_headers(demo_token)
)
assert existing_token_response.status_code == 401, existing_token_response.get_json()
login_after_revoke = client.post(
"/api/auth/login",
json={"username": demo["username"], "password": "demo-original-password"},
)
assert login_after_revoke.status_code == 401, login_after_revoke.get_json()
def test_demo_tenant_visibility_and_role_mutation_gates(client, user_store, login):
super_token = _super_admin_token(user_store, login)
admin = user_store.create_user(
org_id="org-default",
username="visibility-admin",
password="visibility-admin-password",
name="Visibility Admin",
role="admin",
)
user = user_store.create_user(
org_id="org-default",
username="visibility-user",
password="visibility-user-password",
name="Visibility User",
role="user",
)
admin_token = login(admin["username"], "visibility-admin-password")["token"]
user_token = login(user["username"], "visibility-user-password")["token"]
_demo, demo_token = _active_demo_token(
user_store,
username="visibility-demo",
email="visibility-demo@example.com",
)
public_gid, public_pid = _ready_group(
client.application,
org_id="org-default",
creator_id="admin",
visibility="public",
suffix="public",
)
hidden_gid, hidden_pid = _ready_group(
client.application,
org_id="org-default",
creator_id="admin",
visibility="hidden",
suffix="hidden",
)
demo_gid, demo_pid = _ready_group(
client.application,
org_id=Config.DEMO_ORG_ID,
creator_id="admin",
visibility="demo",
suffix="demo",
)
user_groups = client.get("/api/groups", headers=_headers(user_token)).get_json()["groups"]
admin_groups = client.get("/api/groups", headers=_headers(admin_token)).get_json()["groups"]
demo_groups = client.get("/api/groups", headers=_headers(demo_token)).get_json()["groups"]
assert public_gid in {g["id"] for g in user_groups}
assert hidden_gid not in {g["id"] for g in user_groups}
assert demo_gid not in {g["id"] for g in user_groups}
assert {public_gid, hidden_gid} <= {g["id"] for g in admin_groups}
assert demo_gid not in {g["id"] for g in admin_groups}
assert {demo_gid} == {g["id"] for g in demo_groups}
assert client.get(f"/api/groups/{public_gid}", headers=_headers(user_token)).status_code == 200
assert client.get(f"/api/groups/{hidden_gid}", headers=_headers(user_token)).status_code == 403
assert client.get(f"/api/groups/{demo_gid}", headers=_headers(user_token)).status_code == 404
assert client.get(f"/api/groups/{public_gid}", headers=_headers(demo_token)).status_code == 404
assert client.get(f"/api/groups/{demo_gid}", headers=_headers(demo_token)).status_code == 200
demo_create = client.post(
"/api/groups",
json={"product": "Demo must not create"},
headers=_headers(demo_token),
)
demo_persona = client.post(
"/api/me/personas/generate",
json={"mode": "manual", "spec": {}},
headers=_headers(demo_token),
)
user_create = client.post(
"/api/groups",
json={"product": "User private product"},
headers=_headers(user_token),
)
admin_create = client.post(
"/api/groups",
json={"product": "Admin shared product", "visibility": "hidden"},
headers=_headers(admin_token),
)
assert demo_create.status_code == 403, demo_create.get_json()
assert demo_persona.status_code == 403, demo_persona.get_json()
assert user_create.status_code == 201, user_create.get_json()
assert user_create.get_json()["group"]["is_owned"] is True
assert "owner_user_id" not in user_create.get_json()["group"]
assert admin_create.status_code == 201, admin_create.get_json()
assert "owner_user_id" not in admin_create.get_json()["group"]
user_patch = client.patch(
f"/api/groups/{hidden_gid}/visibility",
json={"visibility": "public"},
headers=_headers(user_token),
)
demo_patch = client.patch(
f"/api/groups/{demo_gid}/visibility",
json={"visibility": "public"},
headers=_headers(demo_token),
)
admin_patch = client.patch(
f"/api/groups/{hidden_gid}/visibility",
json={"visibility": "public"},
headers=_headers(admin_token),
)
assert user_patch.status_code == 403, user_patch.get_json()
assert demo_patch.status_code == 403, demo_patch.get_json()
assert admin_patch.status_code == 200, admin_patch.get_json()
assert user_store is not None
assert client.application.extensions["group_store"].get(hidden_gid)["visibility"] == "public"
def test_admin_hidden_group_preview_survives_ready_group_authorization(client, user_store, login):
super_token = _super_admin_token(user_store, login)
admin = user_store.create_user(
org_id="org-default",
username="hidden-preview-admin",
password="hidden-preview-admin-password",
name="Hidden Preview Admin",
role="admin",
)
user = user_store.create_user(
org_id="org-default",
username="hidden-preview-user",
password="hidden-preview-user-password",
name="Hidden Preview User",
role="user",
)
admin_token = login(admin["username"], "hidden-preview-admin-password")["token"]
user_token = login(user["username"], "hidden-preview-user-password")["token"]
gid, pid = _ready_group(
client.application,
org_id="org-default",
creator_id="admin",
visibility="hidden",
suffix="preview",
)
client.application.extensions["llm"] = MockLLM()
admin_start = client.post(
f"/api/chat/{gid}/personas/{pid}/chat/start",
json={"mode": "preview", "scenario": "social"},
headers=_headers(admin_token),
)
user_start = client.post(
f"/api/chat/{gid}/personas/{pid}/chat/start",
json={"mode": "trainee", "scenario": "social"},
headers=_headers(user_token),
)
assert admin_start.status_code == 200, admin_start.get_json()
assert admin_start.get_json()["session"]["mode"] == "preview"
assert user_start.status_code == 403, user_start.get_json()
admin_send = client.post(
f"/api/chat/{gid}/personas/{pid}/chat/send",
json={"text": "Please tell me more."},
headers=_headers(admin_token),
)
assert admin_send.status_code == 200, admin_send.get_json()
assert admin_send.get_json()["session"]["mode"] == "preview"
assert admin_send.get_json()["finished"] is True
assert super_token
def test_demo_can_train_and_read_own_progress_but_cannot_manage_or_analyze(
client, user_store
):
_demo, demo_token = _active_demo_token(
user_store,
username="demo-permission-matrix",
email="demo-permission-matrix@example.com",
)
gid, pid = _ready_group(
client.application,
org_id=Config.DEMO_ORG_ID,
creator_id="demo-permission-matrix",
visibility="demo",
suffix="permission-matrix",
)
client.application.extensions["llm"] = MockLLM()
headers = _headers(demo_token)
started = client.post(
f"/api/chat/{gid}/personas/{pid}/chat/start",
json={"mode": "trainee", "scenario": "social", "locale": "en"},
headers=headers,
)
assert started.status_code == 200, started.get_json()
session_id = started.get_json()["session"]["id"]
sent = client.post(
f"/api/chat/{gid}/personas/{pid}/chat/send",
json={"text": "Tell me more about the price."},
headers=headers,
)
assert sent.status_code == 200, sent.get_json()
board = client.get("/api/me/board", headers=headers)
sessions = client.get("/api/chat/sessions", headers=headers)
session = client.get(f"/api/chat/sessions/{session_id}", headers=headers)
weak_areas = client.get("/api/me/weak-areas", headers=headers)
assert board.status_code == 200, board.get_json()
assert gid in {item["group_id"] for item in board.get_json()["board"]}
assert sessions.status_code == 200, sessions.get_json()
assert session_id in {item["id"] for item in sessions.get_json()["sessions"]}
assert session.status_code == 200, session.get_json()
assert session.get_json()["session"]["id"] == session_id
assert weak_areas.status_code == 200, weak_areas.get_json()
assert "insight" in weak_areas.get_json()
forbidden = (
("post", f"/api/groups/{gid}/personas/{pid}/variant", {}),
("put", f"/api/groups/{gid}/personas/{pid}", {"name": "blocked"}),
("post", f"/api/groups/{gid}/analyze", None),
("post", f"/api/groups/{gid}/reanalyze", None),
("delete", f"/api/groups/{gid}", None),
("get", f"/api/groups/{gid}/report", None),
("get", "/api/analytics", None),
("get", "/api/analytics/export/token", None),
)
for method, path, payload in forbidden:
kwargs = {"headers": headers}
if payload is not None:
kwargs["json"] = payload
response = getattr(client, method)(path, **kwargs)
assert response.status_code == 403, response.get_json()
def test_demo_webhook_redirects_never_downgrade_or_cross_origin():
handler = _HttpsOnlyRedirectHandler()
request = Request("https://hooks.example.test/v1/demo", data=b"secret-body")
for location in (
"http://hooks.example.test/v1/next",
"https://attacker.example.test/collect",
"https://hooks.example.test/v1/next#fragment",
"https://user:password@hooks.example.test/v1/next",
):
with pytest.raises(URLError):
handler.redirect_request(request, None, 302, "Found", {}, location)
redirected = handler.redirect_request(
request, None, 302, "Found", {}, "https://hooks.example.test/v1/next"
)
assert redirected is not None
assert redirected.full_url == "https://hooks.example.test/v1/next"
def test_demo_route_registration_includes_client_visibility_alias(app):
assert any(
rule.rule == "/api/groups/<gid>/visibility" and "PATCH" in rule.methods
for rule in app.url_map.iter_rules()
)
assert any(
rule.rule == "/api/admin/demo-accounts" and "POST" in rule.methods
for rule in app.url_map.iter_rules()
)
assert any(
rule.rule == "/api/admin/demo-accounts/<username>/revoke" and "POST" in rule.methods
for rule in app.url_map.iter_rules()
)