platform/core/users.py

"""Per-user accounts for the platform, persisted in the HF Dataset store (users.json).

Passwords are salted + PBKDF2-HMAC-SHA256 (200k iterations) — never stored or logged in plaintext.
A bootstrap 'admin' account is seeded from APP_PASSWORD so the owner can always log in and create
users; APP_PASSWORD also works as an emergency master for 'admin' if the registry is unreachable.

Each account carries BU access ('all' or a list of team-ids [5=Fisch, 6=Royal]) which drives
allowed_bus() — the basis for per-Business-Unit permissioning (a Royal-only user never sees Fisch).
"""
import os
import hmac
import hashlib
import secrets

import core.store as store

BU_LABELS = {5: 'Fisch', 6: 'Royal'}
_ITER = 200_000

#: Wave 15 C-PERM — the explicit-resolution marker. Mirrors `core.perm_scope.PERMS_VERSION`;
#: kept as a literal here so `users` does not import the permission layer it is read by.
PERMS_VERSION = 1


def _hash(pw, salt):
    return hashlib.pbkdf2_hmac('sha256', str(pw).encode('utf-8'), bytes.fromhex(salt), _ITER).hex()


def _record(pw, name, role, bus, active=True, modules='all', agent=None, email=None,
            perms=None, tenant='royal-imports', platform_admin=False):
    salt = secrets.token_hex(16)
    rec = {'salt': salt, 'hash': _hash(pw, salt), 'name': name, 'role': role,
           'bus': bus, 'active': active, 'modules': modules,
           'agent': agent or None, 'email': email or None,
           # Wave 18 (C1-TENANT, R1): the account's COMPANY. Absent == 'royal-imports' on every
           # pre-wave record — no migration. Login binds the session to THIS value; the posted
           # tenant field can hint but never override it.
           'tenant': str(tenant or 'royal-imports').strip().lower()}
    if platform_admin is True:
        # Wave 19 (R3): the PLATFORM-operator flag — half of `core.platform_admin`'s double lock
        # (the other half is `tenant == 'loopable'`). Written ONLY for True, so every record that
        # is not deliberately promoted keeps its pre-wave shape and answers False by absence.
        # There is no UI writer and there never should be: it is set by provisioning, on purpose.
        rec['platform_admin'] = True
    if perms is not None:
        # Wave 15 C-PERM. A record written WITH perms is migrated by construction — the marker
        # and the block are set together, here, so no writer can create one without the other.
        # (`core.perm_scope` reads an unmarked record as legacy, so a block without its marker
        # would be silently ignored; a marker without a block would deny everything.)
        rec['perms'] = perms
        rec['perms_v'] = PERMS_VERSION
    return rec


def registry():
    return store.get('users')


def ensure_bootstrap():
    """Seed an 'admin' account from APP_PASSWORD ONLY on a truly fresh store (no users file yet).
    Idempotent; no-op if the store is unavailable (the app then falls back to the master-password
    path in verify()).

    Critically, this NEVER overwrites an existing registry: it seeds only when store.exists('users')
    is definitively False. A transient read failure at startup used to return {} and make this
    re-seed just {admin} over the real accounts — that is the bug that wiped users on restart."""
    if not store.available():
        return
    if store.exists('users'):          # present, or uncertain -> never clobber
        return
    try:
        reg = store.get('users', fresh=True)
    except Exception:
        return
    if reg:
        return
    master = os.environ.get('APP_PASSWORD', '')
    if not master:
        return
    try:
        store.put('users', {'admin': _record(master, 'Administrator', 'admin', 'all')})
    except Exception:
        pass


def _public(username, u):
    return {'username': username, 'name': u.get('name', username),
            'role': u.get('role', 'user'), 'bus': u.get('bus', 'all'),
            'modules': u.get('modules', 'all'),
            'agent': u.get('agent'), 'email': u.get('email'),
            # Wave 18 (C1-TENANT): the session's tenant binding travels on the projection or it
            # does not travel — the same rule the perms block states below.
            'tenant': str(u.get('tenant') or 'royal-imports').strip().lower(),
            # Wave 14 C-AVATAR: the profile photo is public-safe by definition (it is served
            # to every grid session via the workspace map); without it here the API session's
            # user record silently drops it and /me can never show your own photo.
            'avatar': u.get('avatar') or None,
            # ⛔ WAVE 15 C-PERM — THE WALL TRAVELS ON THIS PROJECTION OR IT DOES NOT TRAVEL.
            # `deps._user_for` builds every API session from `_public()`, so a `perms` block
            # dropped here is a restricted account served as an unrestricted one — silently, on
            # every route, with nothing to notice. `perms_v` must ride ALONG WITH it and for the
            # same reason inverted: the marker without the block denies everything, the block
            # without the marker is ignored. Two keys, one fact, never separated.
            # `verify_api` asserts a restricted user's SESSION OBJECT carries both, at the mount
            # rather than by grep — a projection is exactly the kind of wiring that looks
            # present in three files and is absent in the one that runs.
            **({'perms': u['perms']} if isinstance(u.get('perms'), dict) else {}),
            **({'perms_v': int(u['perms_v'] or 0)} if u.get('perms_v') else {}),
            # ⛔ WAVE 19 R3 — THE SAME RULE, ON A NEW FIELD. `deps._user_for` builds every API
            # session from this projection, so the platform-admin flag travels here or
            # `core.platform_admin.is_platform_admin(session.user)` is blind and the Loopable
            # admin plane 403s its own operator. Carried ONLY when the record says True, so a
            # session dict for any other account is byte-identical to its pre-wave shape.
            # Not a client leak: `routes_auth._public_user` is a whitelist projection and does
            # not name this key, so it reaches no browser via /login or /me — the client's copy
            # is the separate `platformAdmin` bool on GET /settings, which is derived from this.
            **({'platform_admin': True} if u.get('platform_admin') is True else {}),
            'epoch': int(u.get('epoch') or 0)}


# ------------------------------------------------------------------ session revocation (X3)
# The API's session cookie is SIGNED AND STATELESS: there is no server-side session table to
# delete from, so "log this user out everywhere" needs a number that lives with the account. The
# cookie carries the epoch it was minted under; bumping the account's epoch makes every
# outstanding cookie for that user fail verification on its next use. Absent == 0, so every
# record written before this wave is valid without a migration.
def epoch(username):
    """The current session epoch for `username`. None when there is no such account.

    None is NOT 0. 0 is "this account exists and has never been revoked"; None is "no record" —
    which the session verifier must treat as a reason to refuse, not as a default to compare
    against. (The APP_PASSWORD emergency-master admin has no record at all; the verifier handles
    that case explicitly rather than inventing an epoch for it here.)
    """
    username = (username or '').strip().lower()
    try:
        u = (store.get('users') or {}).get(username)
    except Exception:
        return None
    return int((u or {}).get('epoch') or 0) if u else None


def bump_epoch(username):
    """Revoke every outstanding API session for this account."""
    username = (username or '').strip().lower()

    def _set(reg):
        u = reg.get(username)
        if u:
            u['epoch'] = int(u.get('epoch') or 0) + 1
        return reg
    store.update('users', _set)


def verify(username, pw):
    """Return a public user dict on success, else None. APP_PASSWORD is an emergency master for the
    'admin' login even if the store is unreachable, so the owner is never locked out."""
    username = (username or '').strip().lower()
    if not username or not pw:
        return None
    master = os.environ.get('APP_PASSWORD', '')
    try:
        # read fresh so accounts created moments ago (UI or out-of-band) are recognised at once
        reg = store.get('users', fresh=True)
    except Exception:
        reg = {}
    u = reg.get(username)
    if u is None and '@' in username:
        # Wave 18 (R1): the login box takes a username OR an email — admin@nurilab.id signs in
        # without knowing the slug an admin chose. First case-insensitive email match wins;
        # ambiguity is an admin data problem, not a login feature.
        for k, r in reg.items():
            if isinstance(r, dict) and str(r.get('email') or '').strip().lower() == username:
                username, u = k, r
                break
    if u and u.get('active', True) and hmac.compare_digest(_hash(pw, u['salt']), u['hash']):
        return _public(username, u)
    # emergency master: admin + APP_PASSWORD always works (covers first run / store outage)
    if username == 'admin' and master and hmac.compare_digest(str(pw), master):
        # ⚠ CARRY THE RECORD'S CURRENT EPOCH when there is a record to read, so the session cookie
        # the API mints from this dict AGREES with the stored account.
        #
        # This is not what stops the emergency lockout — `deps._user_for`'s master fallback does
        # that, and a negative control confirmed the lockout is gone with or without this line.
        # What it fixes is subtler and is a SCOPE question: a cookie whose epoch disagrees with the
        # record falls through to that master fallback, which hands back a SYNTHETIC identity
        # (`bus: 'all'`, `modules: 'all'`). An admin whose record narrows either field would
        # therefore be silently WIDENED to consolidated, all-module access for the life of that
        # session. Matching the epoch means the record branch wins and the account's real scope
        # applies, leaving the master fallback as the true last resort it is meant to be.
        # 0 when the store is unreachable, which is the case this branch was written for.
        return {'username': 'admin', 'name': 'Administrator', 'role': 'admin', 'bus': 'all',
                'modules': 'all', 'epoch': int((reg.get('admin') or {}).get('epoch') or 0)}
    return None


def create_user(username, pw, name, role='user', bus='all', modules='all',
                agent=None, email=None, tenant=None, platform_admin=None):
    """Create — or, from app.py's dialog, OVERWRITE — an account.

    ⛔ X3: OVERWRITING AN ACCOUNT MUST NOT RESURRECT ITS OLD SESSIONS. `_record()` builds a fresh
    record with no `epoch` key, i.e. absent == 0. So re-saving an existing username used to reset
    the epoch to 0, and every cookie minted before that account's last password rotation started
    verifying again — a silent un-revocation. `app.py`'s "Add / update a user" calls this function
    for BOTH add and update, so the hole was reachable from the shipped UI.

    Epoch revocation is only ever as strong as the NARROWEST write path that touches the record, so
    the carry-and-bump lives here rather than in each caller: an overwrite is at least as
    session-invalidating as a password change, and it usually IS one.
    """
    username = (username or '').strip().lower()
    if not username or not pw:
        raise ValueError('username and password are required')

    def _add(reg):
        prior = reg.get(username) or {}
        rec = _record(pw, name or username, role, bus, modules=modules,
                      agent=agent, email=email,
                      # An overwrite that names no tenant KEEPS the account's company — a
                      # rename must never quietly move a user between tenants.
                      tenant=(tenant or prior.get('tenant') or 'royal-imports'),
                      # Wave 19 (R3): CARRIED, for the same reason `epoch` is carried below —
                      # `_record` builds a FRESH record, so re-running the provisioner (it is
                      # documented as idempotent) or saving an account through this function
                      # would silently DEMOTE a platform admin and lock the operator out of
                      # their own plane. None = leave as it was; True/False = set it deliberately.
                      platform_admin=(prior.get('platform_admin') is True
                                      if platform_admin is None else platform_admin is True))
        if prior:
            rec['epoch'] = int(prior.get('epoch') or 0) + 1
        reg[username] = rec
        return reg
    store.update('users', _add)


def set_password(username, pw):
    username = (username or '').strip().lower()

    def _set(reg):
        u = reg.get(username)
        if u:
            u['salt'] = secrets.token_hex(16)
            u['hash'] = _hash(pw, u['salt'])
            # X3: a password change revokes every outstanding API session for the account. Bumped
            # INSIDE the same read-modify-write as the hash so the two can never disagree — a
            # separate update() could rotate the password and leave old cookies live if the second
            # write failed.
            u['epoch'] = int(u.get('epoch') or 0) + 1
        return reg
    store.update('users', _set)


def set_active(username, active):
    username = (username or '').strip().lower()

    def _set(reg):
        if username in reg:
            reg[username]['active'] = bool(active)
            # X3: DEACTIVATION must kill live sessions, not just future logins — otherwise a
            # disabled account keeps working until its cookie expires. Bumped on reactivation too:
            # cheap, and it means a re-enabled account never resurrects a stale cookie.
            reg[username]['epoch'] = int(reg[username].get('epoch') or 0) + 1
        return reg
    store.update('users', _set)


def set_platform_admin(username, on):
    """Promote/demote a PLATFORM administrator (wave 19, R3) without touching the password.

    The narrow write, deliberately: `create_user` is the destructive path (fresh salt, fresh
    hash, bumped epoch) and promoting somebody should not sign them out or rotate a credential.
    Cleared by REMOVING the key, so a demoted record goes back to its pre-wave shape rather than
    carrying a `False` that reads as "somebody considered this".

    ⚠ This is the flag only. It grants nothing on its own — `core.platform_admin` also demands
    the `loopable` tenant, and there is no code path anywhere that moves an account between
    tenants, which is what makes the second lock hold.
    """
    username = (username or '').strip().lower()

    def _set(reg):
        u = reg.get(username)
        if u:
            if on is True:
                u['platform_admin'] = True
            else:
                u.pop('platform_admin', None)
        return reg
    store.update('users', _set)


# ------------------------------------------------------------------ activity stamps (wave 19 R4)
# "When did this account last sign in, and is anyone actually using it?" — the two questions the
# Loopable admin plane exists to answer and that NOTHING in the product could answer before this
# wave (there is no login history, no audit log, no request log anywhere).
#
# ⛔ THIS IS THE FIRST HIGH-FREQUENCY WRITER `users.json` HAS EVER HAD, and that bucket also holds
# every password hash, `active`, `epoch` and the permission blocks. Three rules follow, and the
# second one is a correctness rule, not a performance one:
#
#   1. **In-place mutation of ONE key.** Never `_record()`, never a whole-record replace: a stamp
#      that rebuilt the record would reset the salt/hash (locking the user out) or the epoch
#      (silently un-revoking every cookie ever minted for them). `set_password`'s docstring
#      explains why that class of bug is worth naming out loud.
#
#   2. **SYNCHRONOUS FLUSH — deliberately NOT the async path, and this reverses my first draft.**
#      `store.update(flush='async')` rebases on the PROCESS CACHE once a key is `_owned`
#      (`core/store.py:284`) and its worker uploads that whole cached blob. For a
#      table-workspace key, written by one process, that is exactly right. For `users` it is a
#      silent-revert machine: tenant #0's Streamlit host writes the SAME file, so an API process
#      holding a cache from an hour ago would, on its next stamp, upload a blob in which a
#      password rotation or a deactivation performed in the other host simply never happened.
#      A telemetry stamp must not be able to resurrect a disabled account. `flush='sync'` does a
#      FRESH strict read inside the store's lock and then uploads, which is the same discipline
#      every other `users` writer already uses.
#
#   3. **OFF THE REQUEST THREAD, so rule 2 costs nothing.** A sync commit is a hub round-trip, and
#      neither a sign-in nor a random request an hour later should wait for it. Each stamp runs on
#      a short-lived daemon thread; `flush_stamps()` is how a test or a shutdown waits for them.
#      Everything is fail-silent: a stamp is telemetry and may never turn a good login into a
#      failed one — the plane shows an honest "never" instead.
def _now_iso():
    import datetime as _dt
    return _dt.datetime.now(_dt.timezone.utc).isoformat(timespec='seconds')


#: Live stamp threads, so `flush_stamps()` can join them. Bounded by construction — one thread per
#: stamp, and a stamp is at most one per login plus one per account per hour per process.
_STAMPS = []
_STAMPS_LOCK = __import__('threading').Lock()


def _stamp(username, fields):
    """Merge `fields` into ONE account record, on a background thread, with a fresh read."""
    username = (username or '').strip().lower()
    if not username or not fields:
        return None

    def _set(reg):
        u = reg.get(username)
        if isinstance(u, dict):
            u.update(fields)
        return reg

    def _work():
        try:
            store.update('users', _set)          # sync: fresh strict read + blocking upload
        except Exception:
            pass                                  # a lost stamp is a lost stamp, never an error
    import threading
    t = threading.Thread(target=_work, daemon=True, name=f'user-stamp:{username}')
    with _STAMPS_LOCK:
        _STAMPS[:] = [x for x in _STAMPS if x.is_alive()]
        _STAMPS.append(t)
    t.start()
    return t


def flush_stamps(timeout=10.0):
    """Block until outstanding stamp writes have been applied. For gates and shutdown hooks —
    the app never needs it, exactly like `core.store.flush`."""
    with _STAMPS_LOCK:
        pending = list(_STAMPS)
    for t in pending:
        t.join(timeout=timeout)
    return all(not t.is_alive() for t in pending)


def touch_login(username, when=None):
    """Stamp `last_login` (ISO-8601, UTC, OFFSET-BEARING) on a SUCCESSFUL login.

    `username` must be the RESOLVED account key, not what the person typed: `verify()` accepts an
    email address and resolves it to the registry key, so stamping the typed identifier would
    write a stamp onto a key that does not exist and create a phantom account in the registry.

    `last_active` rides along — signing in IS activity, and setting both here means the plane's
    two columns agree the moment somebody logs in rather than an hour later.
    """
    stamp = when or _now_iso()
    return _stamp(username, {'last_login': stamp, 'last_active': stamp})


def touch_active(username, when=None):
    """Stamp `last_active` — "this session did something". Throttled BY THE CALLER (`deps.py`
    holds a process-local last-seen map), so this is not a store round-trip per request."""
    return _stamp(username, {'last_active': when or _now_iso()})


def set_access(username, role=None, bus=None, modules=None, agent=None, email=None,
               name=None, perms=None):
    """Update access fields. agent/email: pass '' to clear, None to leave unchanged —
    the user↔agent link scopes the Customer List page / digests to that agent's book.

    `name` follows the same None-means-unchanged idiom. It is here because it had no setter at all:
    a display name could previously only be changed by re-creating the record through
    `create_user`, i.e. by also resetting the password (and, before the fix above, the session
    epoch). Y4's `PATCH {name?}` needs the narrow write, not the destructive one."""
    username = (username or '').strip().lower()

    def _set(reg):
        u = reg.get(username)
        if u:
            if name is not None:
                u['name'] = name
            if role is not None:
                u['role'] = role
            if bus is not None:
                u['bus'] = bus
            if modules is not None:
                u['modules'] = modules
            if agent is not None:
                u['agent'] = agent or None
            if email is not None:
                u['email'] = email or None
            if perms is not None:
                # Wave 15 C-PERM. Writing perms MIGRATES the record: the marker goes on in the
                # same read-modify-write, so a record can never end up with one and not the
                # other (see `_record`). Whole-block replace, matching the PUT route's shape —
                # a merge would make "remove this restriction" unexpressible.
                u['perms'] = perms
                u['perms_v'] = PERMS_VERSION
        return reg
    store.update('users', _set)


def allowed_bus_labels(user):
    """BU labels this user may select. 'all' -> All+Fisch+Royal; a single BU -> just that BU (no
    'All', so the other BU is never reachable); multiple -> All + each."""
    bus = (user or {}).get('bus', 'all')
    if bus == 'all':
        return ['All', 'Fisch', 'Royal']
    labels = [BU_LABELS[b] for b in bus if b in BU_LABELS]
    if not labels:
        return ['All', 'Fisch', 'Royal']
    return (['All'] + labels) if len(labels) > 1 else labels


def assignable_people(tenant=None):
    """Display names for `user`-typed overlay columns — the tenant's ACTIVE accounts.

    Moved from app.py (2026-07-31) so both hosts serve the same choices. Resolved on every
    call rather than persisted with the column: a snapshot would keep offering people who
    have left and never offer people who joined. Deactivated accounts are excluded; a value
    already stored on a row is untouched — history should still say who owned something.

    Wave 18 (C1-TENANT): pass `tenant` to scope the choices to ONE company — the user registry
    is a global control-plane bucket, and a Nurilab picker offering Royal's staff is a
    cross-tenant name leak. None = unscoped (the Streamlit host, tenant #0's process).
    """
    try:
        reg = registry() or {}
    except Exception:
        return []
    want = str(tenant or '').strip().lower()
    out = []
    for username, u in reg.items():
        if not isinstance(u, dict) or u.get('active') is False:
            continue
        if want and str(u.get('tenant') or 'royal-imports').strip().lower() != want:
            continue
        out.append(str(u.get('name') or username))
    return sorted(set(out))


def set_avatar(username, data_url):
    """Set (or clear, with None/'') the user's profile photo — a data URL (wave 14 C-AVATAR,
    [[loopable-wave14-split]] item 11). Stored VERBATIM; the API route owns validation (mime +
    decoded size) because this value is served back to every grid session. Cleared by removing
    the key, so records without a photo keep their pre-wave shape."""
    username = (username or '').strip().lower()

    def _set(reg):
        u = reg.get(username)
        if u:
            if data_url:
                u['avatar'] = str(data_url)
            else:
                u.pop('avatar', None)
        return reg
    store.update('users', _set)


def avatar_map(tenant=None):
    """{display name -> avatar data URL} for ACTIVE accounts with a photo — the companion of
    `assignable_people()`, keyed by the SAME vocabulary: a `user` cell stores the display
    name, so the display name is the only join a renderer has. Two active accounts sharing a
    display name share one option; the first WITH a photo wins the key rather than a coin
    flip deciding whether the option has a face. `tenant` scopes it exactly as
    `assignable_people(tenant)` does, and for the same leak."""
    try:
        reg = registry() or {}
    except Exception:
        return {}
    want = str(tenant or '').strip().lower()
    out = {}
    for username, u in sorted(reg.items()):
        if not isinstance(u, dict) or u.get('active') is False:
            continue
        if want and str(u.get('tenant') or 'royal-imports').strip().lower() != want:
            continue
        av = u.get('avatar')
        nm = str(u.get('name') or username)
        if av and nm not in out:
            out[nm] = str(av)
    return out