"""HTML views for the read-only restart console (#667). Every interpolated value passes through :func:`_esc`. Values that can carry a filesystem path or free-form operator text additionally pass through :func:`webui.inventory.scrub_text`, which redacts credential-shaped tokens *inside* a string rather than only at its start. The page renders state and never offers a control that would mutate anything: the approval and break-glass panels report authorization and availability, and there is no form, button, or endpoint behind them. """ from __future__ import annotations import html from webui.inventory import scrub_text from webui.restart_console import RestartConsoleSnapshot, SourceStatus def _esc(value: object) -> str: """Escape any value for HTML text or a quoted attribute.""" if value is None: return "" return html.escape(str(value), quote=True) def _esc_text(value: object) -> str: """Escape free-form text after redacting secrets embedded inside it.""" if value is None: return "" return _esc(scrub_text(str(value))) def _bool_badge( value: bool, *, true_label: str = "yes", false_label: str = "no" ) -> str: css = "badge-ok" if value else "badge-blocked" label = true_label if value else false_label return f'{_esc(label)}' def _source_badge(source: SourceStatus) -> str: css = "badge-ok" if source.available else "badge-blocked" badge = f'{_esc(source.status)}' if source.detail: badge += f' {_esc_text(source.detail)}' return badge def _notes_block(snapshot: RestartConsoleSnapshot) -> str: if not snapshot.notes: return "" items = "".join(f"
  • {_esc_text(note)}
  • " for note in snapshot.notes) return f"" def _impact_section(snapshot: RestartConsoleSnapshot) -> str: head = ( "
    " f"

    Impact preview {_source_badge(snapshot.impact_source)}

    " ) impact = snapshot.impact if impact is None: return ( head + "

    No impact preview is available, so the blast " "radius of a restart is unknown. Treat this as unsafe.

    " ) counts = impact.get("counts") or {} verdict = str(impact.get("verdict") or "unknown") verdict_css = "badge-ok" if verdict == "safe" else "badge-blocked" rows = "".join( f"{_esc(key.replace('_', ' '))}{_esc(value)}" for key, value in sorted(counts.items()) ) reasons = "".join( f"
  • {_esc_text(reason)}
  • " for reason in (impact.get("reasons") or []) ) incomplete = "" if not impact.get("inventory_complete", False): detail = "; ".join(str(r) for r in (impact.get("incomplete_reasons") or [])) incomplete = ( "

    Inventory incomplete: " f"{_esc_text(detail or 'unspecified')}. The coordinator fails " "closed on an incomplete inventory.

    " ) sessions = impact.get("affected_sessions") or [] session_rows = "".join( "" f"{_esc(s.get('session_id'))}" f"{_esc(s.get('role'))}" f"{_esc(s.get('pid'))}" f"{_bool_badge(bool(s.get('live')), true_label='live', false_label='idle')}" f"{_bool_badge(not s.get('heartbeat_stale'), true_label='fresh', false_label='stale')}" "" for s in sessions[:50] ) session_table = ( "

    Sessions a restart would terminate

    " "
    " "" "" f"{session_rows}
    SessionRolePIDStateHeartbeat
    " if session_rows else "

    No affected sessions reported.

    " ) truncated = ( f"

    Showing the first 50 of {_esc(len(sessions))} " "affected sessions.

    " if len(sessions) > 50 else "" ) return ( head + "

    Verdict " f"{_esc(verdict)} · " f"blast radius {_esc(impact.get('blast_radius'))} · " f"class {_esc(impact.get('restart_class'))}

    " + incomplete + (f"" if reasons else "") + (f"{rows}
    " if rows else "") + session_table + truncated + "" ) def _drain_section(snapshot: RestartConsoleSnapshot) -> str: head = ( "
    " f"

    Drain proof {_source_badge(snapshot.drain_source)}

    " ) drain = snapshot.drain if drain is None: return ( head + "

    No drain proof has been presented to this view. " "The #661 gate authorizes a restart only against a valid, unexpired, " "clean proof, so the absence of one is a denial, not a pass.

    " "
    " ) reasons = "".join( f"
  • {_esc_text(reason)}
  • " for reason in (drain.get("reasons") or []) ) return ( head + "" f"" f"" f"" f"" f"" "
    Valid{_bool_badge(bool(drain.get('valid')))}
    Clean{_bool_badge(bool(drain.get('clean')))}
    Expired{_bool_badge(not drain.get('expired'), true_label='no', false_label='yes')}
    Tampered{_bool_badge(not drain.get('tampered'), true_label='no', false_label='yes')}
    Proof id{_esc(drain.get('proof_id'))}
    " + (f"" if reasons else "") + "" ) def _reconcile_section(snapshot: RestartConsoleSnapshot) -> str: head = ( "
    " f"

    Post-restart reconcile {_source_badge(snapshot.reconcile_source)}

    " ) proof = snapshot.reconcile if proof is None: return ( head + "

    No post-restart completion proof is recorded. " "Until one is, the last restart's recovery state is unproven.

    " "
    " ) items = "".join( "" f"{_esc(item.get('dimension'))}" f"{_esc(item.get('status'))}" f"{_esc_text(item.get('summary'))}" f"{_bool_badge(not item.get('follow_up_required'), true_label='no', false_label='yes')}" "" for item in (proof.get("items") or []) ) return ( head + "

    Status " f"{_esc(proof.get('overall_status'))} · mode " f"{_esc(proof.get('mode'))} · resolved " f"{_esc(proof.get('resolved_count'))} · unresolved " f"{_esc(proof.get('unresolved_count'))}

    " + ( "
    " "" "" f"{items}
    DimensionStatusSummaryFollow-up required
    " if items else "

    No reconcile dimensions reported.

    " ) + "" ) def _class_matrix_section(snapshot: RestartConsoleSnapshot) -> str: rows = "".join( "" f"{_esc(view.restart_class)}" f"{_esc(view.required_permission)}" f"{_esc(view.expected_blast_radius)}" f"{_esc(view.drain_requirement)}" f"{_esc(view.approval_requirement)}" f"{_bool_badge(view.viewer_may_request)}" f"{_bool_badge(view.viewer_may_execute)}" "" for view in snapshot.restart_classes ) return ( "
    " "

    Restart classes

    " "

    The least-privilege matrix each restart request is " "resolved against. “You may request” and “you may " "execute” are computed for the current viewer role, not for a " "generic operator.

    " "
    " "" "" "" f"{rows}
    ClassPermissionBlast radiusDrainApprovalYou may requestYou may execute
    " "
    " ) def _approval_section(snapshot: RestartConsoleSnapshot) -> str: rows = "".join( "" f"{_esc(a.action_id)}" f"{_esc(a.required_role)}" f"{_bool_badge(a.allowed)}" f"{_bool_badge(a.execution_enabled)}" f"{_esc(a.reason_code)}" f"{_esc_text(a.detail)}" "" for a in snapshot.authorizations ) return ( "
    " "

    Approval controls

    " "

    Authorization is probed the way execution would probe " "it, so “execution enabled” answers whether the action would " "actually run — not merely whether this role outranks the requirement. " "No control on this page performs the action.

    " "
    " "" "" "" f"{rows}
    ActionRequired roleAuthorizedExecution enabledReasonDetail
    " "
    " ) def _break_glass_section(snapshot: RestartConsoleSnapshot) -> str: bg = snapshot.break_glass if not bg.viewer_is_privileged: return ( "
    " "

    Break-glass

    " "

    Break-glass status is visible to operator-class " "roles only. Your role does not carry that authority, so no " "emergency surface is shown.

    " "
    " ) return ( "
    " "

    Break-glass " f"{_bool_badge(bg.available, true_label='available', false_label='unavailable')}" "

    " f"

    {_esc_text(bg.reason)}

    " f"

    Tracked by issue #{_esc(bg.issue)}.

    " "
    " ) def render_restart_console_page(snapshot: RestartConsoleSnapshot) -> str: """Render the whole read-only restart console body.""" return ( "

    Restart status and impact

    " f"

    Generated {_esc(snapshot.generated_at)} · " f"viewer role {_esc(snapshot.viewer_role)} · " f"authenticated {_bool_badge(snapshot.viewer_authenticated)} · " f"read-only {_bool_badge(snapshot.read_only)}

    " + _notes_block(snapshot) + _impact_section(snapshot) + _drain_section(snapshot) + _reconcile_section(snapshot) + _class_matrix_section(snapshot) + _approval_section(snapshot) + _break_glass_section(snapshot) )