"""bw_view_as.py — standard "View As" (impersonation) for BW-Auth (Pattern B) apps.

PROPOSED v2 — merges the "best of both": the current standard drop-in PLUS the
hardening proven in the `with` app (with.bowden.works). Backward compatible with v1 —
every existing call site keeps working; new capabilities are keyword-only args / new
functions. Review target: /srv/system/id-auth/app-auth/. Passes an extended self-check
(python3 bw_view_as_v2.py) that encodes every fix below.

------------------------------------------------------------------------------------
WHAT'S NEW vs v1 (all additive, all optional):

  1. Typed errors      — ViewAsError.code (machine-readable) alongside the message, so
                         SPA apps map codes -> UX; server apps still show the string.
  2. Nesting guard     — start() refuses starting while ALREADY impersonating unless
                         allow_nested=True (v1 silently overwrote). Only behavior change.
  3. Audit hook        — start(..., audit=fn) / stop(..., audit=fn) call
                         fn(action, real, target, mode) so every app gets consistent
                         both-identity audit for free. Audit is best-effort and can
                         NEVER block a stop (the security clear happens first).
  4. Continuous re-auth — verify(session, ...) to call PER-REQUEST in your actor
                         resolver: re-confirms the impersonation is STILL allowed
                         (policy, target validity, no escalation) and fail-closed
                         auto-stops if not. Closes v1's "trusts the flag once set" gap.
  5. Target-validity hook — optional target_valid(target) -> bool, honored by BOTH
                         start() and verify(). This is what lets an app re-check that
                         the target is still a real, current user every request (the
                         thing `with`'s resolve_actor does by re-loading the row) and
                         restores a distinct TARGET_NOT_FOUND code. Without it, a stale
                         target can only be caught by folding existence into
                         can_view_as (see the CONTRACT note below).
  6. SPA contract      — me_fields(session) returns the standard /api/me impersonation
                         block so SPA apps have a first-class helper.

RETAINED from v1 unchanged: report_impersonation wiring (hub oversight), mode='act'
(full act-as) / mode='readonly' default + can_write(), rank_of no-escalation guard,
the two hooks (can_view_as / viewable_targets), banner_html / picker_html / BANNER_CSS.

------------------------------------------------------------------------------------
CONTRACT — the one thing to get right:

    can_view_as(real, target) -> bool

  This is the AUTHORITY for both start() and verify(). It must return True ONLY when
  `real` is *currently* allowed to view as `target`. If your policy is target-existence-
  independent (e.g. "real is an admin"), ALSO pass target_valid= so a deleted/invalid
  target is caught on every request — otherwise a mid-session target deletion would
  leave a "ghost" impersonation (the canonical VIEW-AS.md Example A, `role_of(real) ==
  'admin'`, has exactly this trap; fix it to also assert the target is viewable, or
  pass target_valid). Relationship-scoped hooks ("target in real's clients") already
  fail closed when the relationship or target vanishes, so target_valid is optional
  there.
"""
import html

ACTING_KEY = 'bw_acting_as'
MODE_KEY = 'bw_acting_mode'        # 'readonly' (default) | 'act'


class ViewAsError(Exception):
    """start()/verify() refused — show `.args[0]` to the user, keep the real session.

    NEW in v2: carries a machine-readable `.code` so SPA apps map failures to UX
    without string-matching. Codes are stable:
        NOT_SIGNED_IN | BAD_TARGET | FORBIDDEN | TARGET_NOT_FOUND | ESCALATION | NESTED
    """

    def __init__(self, message, code='FORBIDDEN'):
        super().__init__(message)
        self.code = code


def _norm(session):
    """The real username, normalized (lower/stripped) — the identity used consistently
    for guards AND audit, so enter/exit rows never disagree on the impersonator."""
    return (session.get('user') or '').strip().lower()


def start(session, target, *, can_view_as, rank_of=None, target_valid=None,
          mode='readonly', allow_nested=False, audit=None):
    """Begin impersonation, AUTHORIZED SERVER-SIDE.

    Args:
        session:      your app's session (a mutable mapping).
        target:       the BW username to view as (from the picker — re-checked here).
        can_view_as:  can_view_as(real, target) -> bool. YOUR policy. REQUIRED. See the
                      CONTRACT note in the module docstring.
        rank_of:      optional rank_of(username) -> int. If given, refuses a target of
                      EQUAL or HIGHER rank (no escalation). Strongly recommended.
        target_valid: NEW. optional target_valid(target) -> bool existence/validity
                      check. If given and False, raises TARGET_NOT_FOUND (distinct from
                      the FORBIDDEN policy failure). verify() honors it too.
        mode:         'readonly' (default; writes blocked) or 'act' (real actions).
        allow_nested: NEW. False (default) refuses start() while already impersonating.
                      True restores v1's overwrite AND emits a stop-audit for the
                      abandoned target so the trail stays complete.
        audit:        NEW. optional audit(action, real, target, mode). Called
                      audit('view_as.start', real, target, mode) on success (and
                      'view_as.stop' for an abandoned target under allow_nested).

    Returns the (normalized) target on success. Raises ViewAsError (with .code).
    """
    real = _norm(session)
    if not real:
        raise ViewAsError('You are not signed in.', 'NOT_SIGNED_IN')
    prev = session.get(ACTING_KEY)
    if prev and not allow_nested:
        raise ViewAsError('Exit the current view-as before starting another.', 'NESTED')
    target = (target or '').strip().lower()
    if not target or target == real:
        raise ViewAsError('Pick a different user to view as.', 'BAD_TARGET')
    if not can_view_as(real, target):
        raise ViewAsError('You are not allowed to view as this user.', 'FORBIDDEN')
    if target_valid is not None and not target_valid(target):
        raise ViewAsError('That user no longer exists.', 'TARGET_NOT_FOUND')
    if rank_of is not None and rank_of(target) >= rank_of(real):
        raise ViewAsError('You cannot view as a user with equal or higher privileges.',
                          'ESCALATION')
    if mode not in ('readonly', 'act'):
        mode = 'readonly'
    # Re-pointing a live view-as (opt-in): close the old target's audit trail first.
    if prev and prev != target and audit:
        _safe_audit(audit, 'view_as.stop', real, prev, session.get(MODE_KEY) or 'readonly')
    session[ACTING_KEY] = target
    session[MODE_KEY] = mode
    if audit:
        _safe_audit(audit, 'view_as.start', real, target, mode)
    return target


def stop(session, *, audit=None):
    """Return to the real user. Idempotent, and NEVER blockable: the session is cleared
    BEFORE the (best-effort) audit fires, so a failing audit sink can't leave a user
    stuck impersonating."""
    was = session.get(ACTING_KEY)
    real = _norm(session)
    m = session.get(MODE_KEY) or 'readonly'
    session.pop(ACTING_KEY, None)   # clear FIRST — the security action is unconditional
    session.pop(MODE_KEY, None)
    if audit and was:
        _safe_audit(audit, 'view_as.stop', real, was, m)


def _safe_audit(audit, action, real, target, mode):
    """Call the app's audit sink; swallow its failures. Audit is telemetry — it must
    never break a start/stop or a security auto-stop. (Apps that need hard-guaranteed
    audit should make their sink durable; the drop-in will not sacrifice the clear.)"""
    try:
        audit(action, real, target, mode)
    except Exception:
        pass


def verify(session, *, can_view_as, rank_of=None, target_valid=None, audit=None):
    """NEW in v2. Re-confirm an ACTIVE impersonation is STILL allowed. Call per-request
    in your actor/context resolver — not just at start().

    Re-runs the same authorization start() did (policy + target-validity + no-escalation).
    If the real user's privilege was revoked, the target was deleted, the relationship
    ended, or ranks changed, it FAIL-CLOSED auto-stops and returns False; the request
    then proceeds as the real user. Returns True when the active impersonation is still
    valid, and True (trivially) when not impersonating. Never raises: a raising hook is
    treated as "no longer valid" (fail closed). Never blockable: returns False even if
    the stop's audit sink is down.
    """
    if not is_impersonating(session):
        return True
    real = _norm(session)
    target = (session.get(ACTING_KEY) or '').strip().lower()
    try:
        ok = bool(real) and bool(target) and target != real and can_view_as(real, target)
        if ok and target_valid is not None:
            ok = bool(target_valid(target))
        if ok and rank_of is not None:
            ok = rank_of(target) < rank_of(real)
    except Exception:
        ok = False                       # fail closed — can't confirm ⇒ drop
    if not ok:
        stop(session, audit=audit)       # _safe_audit inside ⇒ never blockable
        return False
    return True


def real_username(session):
    """The actually-signed-in BW user — owns the 'return' control + audit."""
    return session.get('user')


def effective_username(session):
    """The user the app should act as for DATA + PERMISSIONS (the impersonated user
    while viewing-as, else the real user). Use THIS everywhere in app logic."""
    return session.get(ACTING_KEY) or session.get('user')


def is_impersonating(session):
    return bool(session.get(ACTING_KEY))


def mode(session):
    return session.get(MODE_KEY) or 'readonly'


def can_write(session):
    """False while impersonating in read-only mode — gate your write/POST handlers on
    this so a 'view as' can't accidentally take actions as someone else."""
    return (not is_impersonating(session)) or session.get(MODE_KEY) == 'act'


def me_fields(session, *, label=None):
    """NEW in v2. The standard impersonation block for an SPA's /api/me response, so
    every SPA app exposes the same shape (the client draws the banner from it):

        {"impersonating": bool, "viewing_as": str|None, "viewing_as_label": str|None,
         "real_user": str|None, "can_write": bool, "mode": "readonly"|"act"}

    `label` optionally maps the effective username to a display name."""
    active = is_impersonating(session)
    who = session.get(ACTING_KEY) if active else None
    return {
        'impersonating': active,
        'viewing_as': who,
        'viewing_as_label': (label(who) if (label and who) else who),
        'real_user': session.get('user'),
        'can_write': can_write(session),
        'mode': mode(session),
    }


# --- the standard, un-hideable UI (theme-consistent; override the CSS if you like) ---

BANNER_CSS = """
.bw-viewas-banner{position:fixed;left:0;right:0;bottom:0;z-index:2147483647;
 font-family:system-ui,-apple-system,sans-serif;font-size:14px;padding:10px 16px;
 text-align:center;color:#1f2937;background:#fde68a;border-top:2px solid #d97706;
 box-shadow:0 -2px 10px rgba(0,0,0,.18)}
.bw-viewas-banner a{color:#92400e;font-weight:700;text-decoration:underline;margin-left:10px}
.bw-viewas-banner .ro{opacity:.75;font-weight:600;margin-left:6px}
.bw-viewas-picker{display:inline-flex;gap:8px;align-items:center;font:inherit}
body:has(.bw-viewas-banner){padding-bottom:52px}  /* keep the banner off the content */
"""


def banner_html(session, *, stop_url='/view-as/stop', label=None):
    """The always-visible impersonation banner (render it on every page). Returns ''
    when not impersonating. `label` optionally maps a username to a display name."""
    if not is_impersonating(session):
        return ''
    who = session.get(ACTING_KEY)
    shown = html.escape(str(label(who) if label else who))
    ro = '<span class="ro">(read-only)</span>' if mode(session) != 'act' else ''
    return (f'<div class="bw-viewas-banner" role="status">Viewing as <b>{shown}</b> {ro}'
            f'<a href="{html.escape(stop_url)}">Return to yourself</a></div>')


def picker_html(targets, *, start_url='/view-as/start', csrf=None):
    """A minimal 'View As' picker built from viewable_targets(real). Render it only for
    a user allowed to impersonate (your template decides). Each target is a dict
    {'username': ..., 'label': ...}. POSTs {target} (+ optional csrf) to start_url."""
    if not targets:
        return ''
    opts = ''.join(
        f'<option value="{html.escape(t["username"])}">'
        f'{html.escape(t.get("label") or t["username"])}</option>' for t in targets)
    csrf_field = (f'<input type="hidden" name="csrf" value="{html.escape(csrf)}">'
                  if csrf else '')
    return (f'<form class="bw-viewas-picker" method="POST" action="{html.escape(start_url)}">'
            f'{csrf_field}<label>View as: <select name="target">{opts}</select></label>'
            f'<button type="submit">Go</button></form>')


# Tiny self-check: the mechanism + guards, no I/O. Extends v1 and encodes every v2 fix.
def _selfcheck():
    allow = lambda real, tgt: True
    rank = {'rian': 100, 'adi': 50, 'jane': 10}.get

    # --- v1 parity ---
    s = {'user': 'rian'}
    start(s, 'jane', can_view_as=allow, rank_of=rank)
    assert effective_username(s) == 'jane' and real_username(s) == 'rian'
    assert is_impersonating(s) and not can_write(s)          # read-only default
    stop(s); assert effective_username(s) == 'rian' and not is_impersonating(s)
    start(s, 'adi', can_view_as=allow, rank_of=rank, mode='act'); assert can_write(s)
    stop(s)
    for bad, code in [('rian', 'BAD_TARGET'), ('', 'BAD_TARGET')]:
        try:
            start(s, bad, can_view_as=allow); raise AssertionError('accepted')
        except ViewAsError as e:
            assert e.code == code
    try:
        start({'user': 'adi'}, 'rian', can_view_as=allow, rank_of=rank)
    except ViewAsError as e:
        assert e.code == 'ESCALATION'
    try:
        start(s, 'jane', can_view_as=lambda r, t: False)
    except ViewAsError as e:
        assert e.code == 'FORBIDDEN'
    try:
        start({}, 'x', can_view_as=allow)
    except ViewAsError as e:
        assert e.code == 'NOT_SIGNED_IN'

    # --- #2 nesting guard + #5 abandoned-target stop-audit on re-point ---
    seen = []
    audit = lambda a, r, t, m: seen.append((a, r, t, m))
    start(s, 'jane', can_view_as=allow)
    try:
        start(s, 'adi', can_view_as=allow); raise AssertionError('allowed nest')
    except ViewAsError as e:
        assert e.code == 'NESTED'
    seen.clear()
    assert start(s, 'adi', can_view_as=allow, allow_nested=True, audit=audit) == 'adi'
    assert seen == [('view_as.stop', 'rian', 'jane', 'readonly'),
                    ('view_as.start', 'rian', 'adi', 'readonly')], seen
    stop(s)

    # --- #3 audit hook: both-identity, and #4 consistent identity enter==exit ---
    seen.clear()
    s2 = {'user': 'Rian'}                                     # mixed-case on purpose
    start(s2, 'jane', can_view_as=allow, mode='act', audit=audit)
    stop(s2, audit=audit)
    assert seen == [('view_as.start', 'rian', 'jane', 'act'),
                    ('view_as.stop', 'rian', 'jane', 'act')], seen   # real agrees
    seen.clear(); stop(s2, audit=audit); assert seen == []   # no-op stop stays silent

    # --- #2/#3 a FAILING audit sink can NEVER block a stop (security clear first) ---
    boom = lambda *a: (_ for _ in ()).throw(RuntimeError('sink down'))
    start(s, 'jane', can_view_as=allow)
    stop(s, audit=boom)                                       # must not raise
    assert not is_impersonating(s)                            # cleared despite bad sink

    # --- #4/verify: fail-closed auto-stop, unblockable even if audit throws ---
    start(s, 'jane', can_view_as=allow)
    assert verify(s, can_view_as=allow) is True and is_impersonating(s)
    assert verify(s, can_view_as=lambda r, t: False, audit=boom) is False
    assert not is_impersonating(s)                            # dropped despite bad sink
    assert verify(s, can_view_as=allow) is True               # trivially true when idle

    # --- #1 HIGH: deleted target is dropped by verify via target_valid ---
    start(s, 'jane', can_view_as=allow)                      # policy: real-is-admin style
    exists = {'jane'}                                         # ...then jane is deleted:
    exists.discard('jane')
    assert verify(s, can_view_as=allow, target_valid=lambda t: t in exists) is False
    assert not is_impersonating(s)                            # no ghost impersonation
    # and start() surfaces it as a distinct code
    try:
        start(s, 'ghost', can_view_as=allow, target_valid=lambda t: False)
    except ViewAsError as e:
        assert e.code == 'TARGET_NOT_FOUND'

    # --- verify catches a mid-session escalation (rank change) ---
    start(s, 'jane', can_view_as=allow, rank_of=rank)
    demoted = {'rian': 5, 'jane': 10}.get
    assert verify(s, can_view_as=allow, rank_of=demoted) is False
    assert not is_impersonating(s)

    # --- #6 a RAISING policy hook at verify fails closed (drops), never propagates ---
    start(s, 'jane', can_view_as=allow)
    kaboom = lambda r, t: (_ for _ in ()).throw(RuntimeError('db down'))
    assert verify(s, can_view_as=kaboom) is False
    assert not is_impersonating(s)

    # --- me_fields SPA contract ---
    start(s, 'jane', can_view_as=allow, mode='act')
    assert me_fields(s, label={'jane': 'Jane Q'}.get) == {
        'impersonating': True, 'viewing_as': 'jane', 'viewing_as_label': 'Jane Q',
        'real_user': 'rian', 'can_write': True, 'mode': 'act'}
    stop(s); assert me_fields(s)['impersonating'] is False

    print('bw_view_as v2 self-check OK: v1 parity + typed errors + nesting guard + '
          'consistent/unblockable audit + fail-closed verify + target_valid '
          '(no ghost impersonation) + me_fields SPA contract')


if __name__ == '__main__':
    _selfcheck()