Add sudo OTP verification for destructive operations.

Gate superuser promote/demote, API node deletion, and account
deletion behind an 8-digit email OTP confirmation step.

remarkbox/api/remarkbox_client.py

@@ -148,17 +148,20 @@ class RemarkboxClient:
         if self._cookie_file and hasattr(self._cookie_jar, "save"):
             self._cookie_jar.save(ignore_discard=True, ignore_expires=True)
 
-    def _request(self, method, path, body=None):
+    def _request(self, method, path, body=None, headers=None):
         """Make an HTTP request and return parsed JSON."""
         url = self.url + path
         data = None
-        headers = {}
+        _headers = {}
 
         if body is not None:
             data = json.dumps(body).encode("utf-8")
-            headers["Content-Type"] = "application/json"
+            _headers["Content-Type"] = "application/json"
 
-        req = urllib.request.Request(url, data=data, headers=headers, method=method)
+        if headers:
+            _headers.update(headers)
+
+        req = urllib.request.Request(url, data=data, headers=_headers, method=method)
 
         try:
             resp = self._opener.open(req)
@@ -351,16 +354,23 @@ class RemarkboxClient:
         """
         return self._request("PATCH", "/api/v1/nodes/{}".format(node_id), {"locked": False})
 
-    def delete_node(self, node_id):
-        """Delete a node permanently (requires moderator).
+    def delete_node(self, node_id, sudo_otp=None):
+        """Delete a node permanently (requires moderator + sudo OTP).
+
+        First call without sudo_otp returns 202 and emails a code.
+        Second call with the code performs the deletion.
 
         Args:
             node_id: UUID of the node to delete
+            sudo_otp: 8-digit confirmation code from email (optional)
 
         Returns:
-            dict with key: deleted (the node ID)
+            dict with key: deleted (the node ID), or status/message if OTP required
         """
-        return self._request("DELETE", "/api/v1/nodes/{}".format(node_id))
+        headers = {}
+        if sudo_otp:
+            headers["X-Sudo-OTP"] = sudo_otp
+        return self._request("DELETE", "/api/v1/nodes/{}".format(node_id), headers=headers)
 
     # ----- Auth -----
 

remarkbox/api/views.py

@@ -24,7 +24,8 @@ from remarkbox.models.namespace import (
 )
 from remarkbox.models.uri import get_or_create_uri
 from remarkbox.models.node import Node, get_or_create_node_by_uri
-from remarkbox.lib.mail import send_verification_digits_to_email
+from remarkbox.lib.mail import send_verification_digits_to_email, send_sudo_otp_email
+from remarkbox.models.sudo_otp import create_sudo_otp, verify_sudo_otp
 from remarkbox.lib.notify import schedule_notifications
 from remarkbox.views import verify_pending_nodes_in_session
 
@@ -864,7 +865,7 @@ def api_edit_node(request):
     require_csrf=False,
 )
 def api_delete_node(request):
-    """Delete a node permanently (requires moderator)."""
+    """Delete a node permanently (requires moderator + sudo OTP)."""
     node_id = request.matchdict["node_id"]
     node = get_node_by_id(request.dbsession, node_id)
 
@@ -886,6 +887,31 @@ def api_delete_node(request):
         request.response.status_code = 403
         return {"error": "Only moderators can delete nodes"}
 
+    # Sudo OTP gate.
+    sudo_otp = request.headers.get("X-Sudo-OTP", "").strip()
+    action_key = "delete_node:{}".format(node_id)
+
+    if not sudo_otp:
+        action_description = "Delete node {}".format(node_id)
+        code = create_sudo_otp(
+            request.dbsession,
+            action_key,
+            action_description,
+            client_ip=str(request.client_addr),
+        )
+        send_sudo_otp_email(request, request.user.email, action_description, code)
+        request.response.status_code = 202
+        return {
+            "status": "otp_required",
+            "message": "A confirmation code has been sent to your email. "
+                       "Repeat this request with the X-Sudo-OTP header.",
+        }
+
+    ok, err = verify_sudo_otp(request.dbsession, action_key, sudo_otp)
+    if not ok:
+        request.response.status_code = 403
+        return {"error": err}
+
     node_id_str = str(node.id)
     request.dbsession.delete(node)
     request.dbsession.flush()

remarkbox/lib/mail.py

@@ -13,6 +13,8 @@ from remarkbox.lib.mail_messages import (
     WELCOME_2_TEXT,
     WELCOME_2_HTML,
     OPERATOR_HTML,
+    SUDO_OTP_TEXT,
+    SUDO_OTP_HTML,
 )
 
 import dkim
@@ -197,6 +199,14 @@ def send_verification_digits_to_email(request, to_email, raw_digits):
     send_pyramid_email(request, to_email, subject, message_text, message_html)
 
 
+def send_sudo_otp_email(request, to_email, action_description, code):
+    """Send a sudo OTP confirmation email."""
+    subject = "Security Verification - {}".format(code)
+    message_text = SUDO_OTP_TEXT.format(action=action_description, code=code)
+    message_html = SUDO_OTP_HTML.format(action=action_description, code=code)
+    send_pyramid_email(request, to_email, subject, message_text, message_html)
+
+
 def send_operator_email(request, msg):
     send_pyramid_email(
         request,

remarkbox/lib/mail_messages.py

@@ -117,6 +117,50 @@ WELCOME_2_HTML = """
 </html>
 """
 
+SUDO_OTP_TEXT = """
+Security Verification Required
+
+Action: {action}
+
+Your one-time confirmation code is:
+
+ \n{code}\n
+
+This code expires in 15 minutes.
+
+If you did not request this action, you can safely ignore this email.
+"""
+
+SUDO_OTP_HTML = """
+<!DOCTYPE html>
+<html>
+<head>
+<title>Security Verification</title>
+<style>
+  .otp-code {{
+    font-size: 3em;
+    font-weight: bold;
+    letter-spacing: 0.1em;
+    margin: 1em 0;
+  }}
+</style>
+</head>
+  <body>
+    <h2>Security Verification Required</h2>
+
+    <p><strong>Action:</strong> {action}</p>
+
+    <p>Your one-time confirmation code is:</p>
+
+    <h1 class="otp-code">{code}</h1>
+
+    <p>This code expires in 15 minutes.</p>
+
+    <p>If you did not request this action, you can safely ignore this email.</p>
+  </body>
+</html>
+"""
+
 OPERATOR_HTML = """<!DOCTYPE html>
 <html>
 <head>

remarkbox/lib/sudo.py

@@ -0,0 +1,14 @@
+from remarkbox.models.sudo_otp import create_sudo_otp
+from remarkbox.lib.mail import send_sudo_otp_email
+
+
+def request_sudo_otp(request, action_key, action_description):
+    """Generate a sudo OTP code, email it to the current user, and return the code."""
+    code = create_sudo_otp(
+        request.dbsession,
+        action_key,
+        action_description,
+        client_ip=str(request.client_addr),
+    )
+    send_sudo_otp_email(request, request.user.email, action_description, code)
+    return code

remarkbox/models/__init__.py

@@ -20,6 +20,7 @@ from .notification import *
 from .pay_what_you_can import *
 from .payment import *
 from .webmention import *
+from .sudo_otp import *
 
 # run configure_mappers after defining all of the models to ensure
 # all relationships can be setup

remarkbox/models/sudo_otp.py

@@ -0,0 +1,90 @@
+import os
+import time
+
+from sqlalchemy import Column, Unicode, BigInteger
+
+from .meta import Base
+
+
+class SudoOtp(Base):
+    """One-time password for gating destructive operations."""
+
+    __tablename__ = "rb_sudo_otp"
+
+    action_key = Column(Unicode(256), primary_key=True)
+    code = Column(Unicode(8), nullable=False)
+    action = Column(Unicode(512), nullable=False)
+    client_ip = Column(Unicode(45), nullable=True)
+    created_at = Column(BigInteger, nullable=False)
+    expires_at = Column(BigInteger, nullable=False)
+
+
+# TTL in milliseconds (15 minutes).
+SUDO_OTP_TTL_MS = 15 * 60 * 1000
+
+
+def generate_sudo_otp_code():
+    """Return a zero-padded 8-digit numeric string."""
+    n = int.from_bytes(os.urandom(5), "big") % (10**8)
+    return str(n).zfill(8)
+
+
+def create_sudo_otp(dbsession, action_key, action, client_ip=None):
+    """Create or replace a sudo OTP for the given action_key.
+
+    Opportunistically cleans up expired rows.
+    Returns the raw 8-digit code string.
+    """
+    now_ms = int(time.time() * 1000)
+
+    # Clean up expired rows (best-effort, small table).
+    dbsession.query(SudoOtp).filter(SudoOtp.expires_at < now_ms).delete()
+
+    code = generate_sudo_otp_code()
+
+    existing = dbsession.query(SudoOtp).get(action_key)
+    if existing:
+        existing.code = code
+        existing.action = action
+        existing.client_ip = client_ip
+        existing.created_at = now_ms
+        existing.expires_at = now_ms + SUDO_OTP_TTL_MS
+    else:
+        otp = SudoOtp(
+            action_key=action_key,
+            code=code,
+            action=action,
+            client_ip=client_ip,
+            created_at=now_ms,
+            expires_at=now_ms + SUDO_OTP_TTL_MS,
+        )
+        dbsession.add(otp)
+
+    dbsession.flush()
+    return code
+
+
+def verify_sudo_otp(dbsession, action_key, code):
+    """Verify a sudo OTP.
+
+    Returns (True, None) on success or (False, error_message) on failure.
+    Deletes the OTP on success (single-use).
+    """
+    now_ms = int(time.time() * 1000)
+
+    otp = dbsession.query(SudoOtp).get(action_key)
+    if otp is None:
+        return False, "No OTP found. Please request a new code."
+
+    if otp.expires_at < now_ms:
+        dbsession.delete(otp)
+        dbsession.flush()
+        return False, "OTP has expired. Please request a new code."
+
+    if otp.code != code:
+        return False, "Invalid OTP code."
+
+    # Success — single-use, delete it.
+    dbsession.delete(otp)
+    dbsession.flush()
+    return True, None

remarkbox/scripts/alembic/versions/7c624a8fae9e_create_rb_sudo_otp_table.py

@@ -0,0 +1,32 @@
+"""create rb_sudo_otp table
+
+Revision ID: 7c624a8fae9e
+Revises: e99b1524710c
+Create Date: 2026-02-05 16:06:35.820072
+
+"""
+
+# revision identifiers, used by Alembic.
+revision = '7c624a8fae9e'
+down_revision = 'e99b1524710c'
+branch_labels = None
+depends_on = None
+
+from alembic import op
+import sqlalchemy as sa
+
+
+def upgrade():
+    op.create_table('rb_sudo_otp',
+        sa.Column('action_key', sa.Unicode(length=256), nullable=False),
+        sa.Column('code', sa.Unicode(length=8), nullable=False),
+        sa.Column('action', sa.Unicode(length=512), nullable=False),
+        sa.Column('client_ip', sa.Unicode(length=45), nullable=True),
+        sa.Column('created_at', sa.BigInteger(), nullable=False),
+        sa.Column('expires_at', sa.BigInteger(), nullable=False),
+        sa.PrimaryKeyConstraint('action_key'),
+    )
+
+
+def downgrade():
+    op.drop_table('rb_sudo_otp')

remarkbox/templates/confirm-delete-account.j2

@@ -4,6 +4,30 @@
 
 <h4>{{ the_title }}</h4>
 
+{% if otp_requested %}
+
+<p>
+An 8-digit confirmation code has been sent to <b>{{ request.user.email }}</b>.
+Enter it below to permanently delete your account.
+</p>
+
+<br>
+
+<form method="post" action="{{ request.link_prefix }}/u/delete-account">
+{% include 'snippets/csrf.j2' %}
+<label for="sudo-otp">Confirmation code:</label>
+<input type="text" name="sudo_otp" id="sudo-otp" class="common-text-input"
+       pattern="[0-9]{8}" inputmode="numeric" maxlength="8"
+       placeholder="12345678" autocomplete="off" autofocus required>
+<br>
+<br>
+{% set submit_button_value = 'permanently delete my account' %}
+{% set submit_button_classes = 'button-right' %}
+{% include 'snippets/submit.j2' %}
+</form>
+
+{% else %}
+
 <p>
 This action is <b>permanent and cannot be undone</b>. Deleting your account will:
 </p>
@@ -31,6 +55,8 @@ Your comments will remain visible but will no longer be associated with your acc
 {% include 'snippets/submit.j2' %}
 </form>
 
+{% endif %}
+
 <br>
 <a href="{{ request.link_prefix }}/u/settings" class="button button-right">cancel</a>
 

remarkbox/templates/confirm-sudo-otp.j2

@@ -0,0 +1,35 @@
+{% extends request.base_template -%}
+{% block title %}Security Verification | {{ request.domain }}{%- endblock -%}
+{% block content -%}
+
+<h4>Security Verification Required</h4>
+
+<p>
+<strong>Action:</strong> {{ action_description }}
+</p>
+
+<p>
+An 8-digit confirmation code has been sent to <b>{{ request.user.email }}</b>.
+Enter it below to proceed.
+</p>
+
+<br>
+
+<form method="post" action="{{ confirm_url }}">
+{% include 'snippets/csrf.j2' %}
+<input type="hidden" name="email" value="{{ email }}" />
+<label for="sudo-otp">Confirmation code:</label>
+<input type="text" name="sudo_otp" id="sudo-otp" class="common-text-input"
+       pattern="[0-9]{8}" inputmode="numeric" maxlength="8"
+       placeholder="12345678" autocomplete="off" autofocus required>
+<br>
+<br>
+{% set submit_button_value = 'confirm' %}
+{% set submit_button_classes = 'button-right' %}
+{% include 'snippets/submit.j2' %}
+</form>
+
+<br>
+<a href="{{ request.route_url('topsecret-users') }}" class="button button-right">cancel</a>
+
+{%- endblock -%}

remarkbox/views/authenticated/authenticated.py

@@ -20,6 +20,9 @@ from remarkbox.views import get_referer_or_home, user_required, reject_stand_alo
 
 from remarkbox.scripts.modify_namespace import delete_namespace_cascade
 
+from remarkbox.lib.sudo import request_sudo_otp
+from remarkbox.models.sudo_otp import verify_sudo_otp
+
 from string import digits
 
 try:
@@ -566,16 +569,31 @@ def namespace_dump_to_json(request):
 @view_config(route_name="embed-user-delete-account", renderer="confirm-delete-account.j2")
 @user_required()
 def delete_account(request):
-    """Delete the authenticated user's account after confirmation."""
+    """Delete the authenticated user's account after OTP confirmation."""
     if request.method == "POST":
-        confirm = request.params.get("confirm-delete", "")
-        if confirm == "DELETE":
+        sudo_otp = request.params.get("sudo_otp", "").strip()
+        action_key = "delete_account:{}".format(request.user.email)
+
+        if sudo_otp:
+            # Step 2: verify OTP and execute deletion.
+            ok, err = verify_sudo_otp(request.dbsession, action_key, sudo_otp)
+            if not ok:
+                request.session.flash((err, "error"))
+                return {"the_title": "Delete My Account", "otp_requested": True}
+
             request.user.anonymize_account()
             request.session.flash(
                 ("Your account has been deleted and your data has been anonymized.", "success")
             )
             request.session["authenticated_user_id"] = None
             return HTTPFound("/")
+
+        confirm = request.params.get("confirm-delete", "")
+        if confirm == "DELETE":
+            # Step 1: generate OTP and show OTP input.
+            action_description = "Delete account {}".format(request.user.email)
+            request_sudo_otp(request, action_key, action_description)
+            return {"the_title": "Delete My Account", "otp_requested": True}
         else:
             request.session.flash(
                 ('You must type "DELETE" to confirm account deletion.', "error")

remarkbox/views/authenticated/topsecret.py

@@ -9,7 +9,9 @@ from remarkbox.models import (
     get_topsecret_nodes,
 )
 from remarkbox.models.user import get_user_by_email, User
+from remarkbox.models.sudo_otp import verify_sudo_otp
 
+from remarkbox.lib.sudo import request_sudo_otp
 from remarkbox.views import super_fly_required
 
 
@@ -94,41 +96,97 @@ def topsecret_users(request):
     }
 
 
-@view_config(route_name="topsecret-user-promote", request_method="POST")
+@view_config(
+    route_name="topsecret-user-promote",
+    request_method="POST",
+    renderer="confirm-sudo-otp.j2",
+)
 @super_fly_required
 def topsecret_user_promote(request):
     email = request.params.get("email", "").strip().lower()
-    if email:
-        user = get_user_by_email(request.dbsession, email)
-        if user:
-            user.is_superuser = True
-            request.dbsession.add(user)
-            request.dbsession.flush()
-            request.session.flash(
-                ("Promoted {} ({}) to superuser.".format(user.name, user.email), "success")
-            )
-        else:
-            request.session.flash(
-                ("No user found with email: {}".format(email), "error")
-            )
+    sudo_otp = request.params.get("sudo_otp", "").strip()
+
+    if not email:
+        return HTTPFound(request.route_url("topsecret-users"))
+
+    user = get_user_by_email(request.dbsession, email)
+    if not user:
+        request.session.flash(
+            ("No user found with email: {}".format(email), "error")
+        )
+        return HTTPFound(request.route_url("topsecret-users"))
+
+    action_key = "promote:{}".format(email)
+    action_description = "Promote {} ({}) to superuser".format(user.name, email)
+
+    if not sudo_otp:
+        # Step 1: generate OTP and show confirmation form.
+        request_sudo_otp(request, action_key, action_description)
+        return {
+            "the_title": "Confirm Promote",
+            "action_description": action_description,
+            "email": email,
+            "confirm_url": request.route_url("topsecret-user-promote"),
+        }
+
+    # Step 2: verify OTP and execute.
+    ok, err = verify_sudo_otp(request.dbsession, action_key, sudo_otp)
+    if not ok:
+        request.session.flash((err, "error"))
+        return HTTPFound(request.route_url("topsecret-users"))
+
+    user.is_superuser = True
+    request.dbsession.add(user)
+    request.dbsession.flush()
+    request.session.flash(
+        ("Promoted {} ({}) to superuser.".format(user.name, user.email), "success")
+    )
     return HTTPFound(request.route_url("topsecret-users"))
 
 
-@view_config(route_name="topsecret-user-demote", request_method="POST")
+@view_config(
+    route_name="topsecret-user-demote",
+    request_method="POST",
+    renderer="confirm-sudo-otp.j2",
+)
 @super_fly_required
 def topsecret_user_demote(request):
     email = request.params.get("email", "").strip().lower()
-    if email:
-        user = get_user_by_email(request.dbsession, email)
-        if user:
-            user.is_superuser = False
-            request.dbsession.add(user)
-            request.dbsession.flush()
-            request.session.flash(
-                ("Demoted {} ({}) from superuser.".format(user.name, user.email), "success")
-            )
-        else:
-            request.session.flash(
-                ("No user found with email: {}".format(email), "error")
-            )
+    sudo_otp = request.params.get("sudo_otp", "").strip()
+
+    if not email:
+        return HTTPFound(request.route_url("topsecret-users"))
+
+    user = get_user_by_email(request.dbsession, email)
+    if not user:
+        request.session.flash(
+            ("No user found with email: {}".format(email), "error")
+        )
+        return HTTPFound(request.route_url("topsecret-users"))
+
+    action_key = "demote:{}".format(email)
+    action_description = "Demote {} ({}) from superuser".format(user.name, email)
+
+    if not sudo_otp:
+        # Step 1: generate OTP and show confirmation form.
+        request_sudo_otp(request, action_key, action_description)
+        return {
+            "the_title": "Confirm Demote",
+            "action_description": action_description,
+            "email": email,
+            "confirm_url": request.route_url("topsecret-user-demote"),
+        }
+
+    # Step 2: verify OTP and execute.
+    ok, err = verify_sudo_otp(request.dbsession, action_key, sudo_otp)
+    if not ok:
+        request.session.flash((err, "error"))
+        return HTTPFound(request.route_url("topsecret-users"))
+
+    user.is_superuser = False
+    request.dbsession.add(user)
+    request.dbsession.flush()
+    request.session.flash(
+        ("Demoted {} ({}) from superuser.".format(user.name, user.email), "success")
+    )
     return HTTPFound(request.route_url("topsecret-users"))