Harden admin user email handling

.gitignore

@@ -13,6 +13,10 @@ __pycache__/
 .pytest_cache/
 .hypothesis/
 
+# Local Python virtual environments
+.venv/
+.venv-*/
+
 # BenchPark repository (external reference)
 benchpark/
 

result_server/routes/admin.py

@@ -17,7 +17,7 @@
     url_for,
 )
 
-from utils.admin_policy import parse_affiliations
+from utils.admin_policy import is_valid_email, parse_affiliations
 from utils.audit_logging import audit_event
 from utils.rate_limit import rate_limited
 from utils.user_store import get_user_store
@@ -121,6 +121,17 @@ def add_user():
     if not email:
         flash("Email is required.")
         return redirect(url_for("admin.users"))
+    if not is_valid_email(email):
+        audit_event(
+            "admin_user_invite_rejected",
+            actor=session.get("user_email"),
+            target=email[:64],
+            result="failure",
+            level=logging.WARNING,
+            details={"reason": "invalid_email_format"},
+        )
+        flash("Invalid email address.")
+        return redirect(url_for("admin.users"))
 
     if store.user_exists(email):
         flash(f"User {email} is already registered. Use 'Reinvite' to reset their TOTP.")

result_server/templates/admin_users.html

@@ -128,13 +128,13 @@ <h2 class="section-title">Registered Users</h2>
                                 <button type="submit" class="btn btn-warning">Update</button>
                             </form>
                             {% if u.email != session.get('user_email') %}
-                            <form class="inline-form" method="POST" action="{{ url_for('admin.reinvite_user', email=u.email) }}"
-                                  onsubmit="return confirm('Reset TOTP for {{ u.email }}?');">
+                            <form class="inline-form js-confirm" method="POST" action="{{ url_for('admin.reinvite_user', email=u.email) }}"
+                                  data-confirm-message="Reset TOTP for {{ u.email }}?">
                                 {% if csrf_token is defined %}<input type="hidden" name="csrf_token" value="{{ csrf_token() }}">{% endif %}
                                 <button type="submit" class="btn btn-primary">Reinvite</button>
                             </form>
-                            <form class="inline-form" method="POST" action="{{ url_for('admin.delete_user', email=u.email) }}"
-                                  onsubmit="return confirm('Delete {{ u.email }}?');">
+                            <form class="inline-form js-confirm" method="POST" action="{{ url_for('admin.delete_user', email=u.email) }}"
+                                  data-confirm-message="Delete {{ u.email }}?">
                                 {% if csrf_token is defined %}<input type="hidden" name="csrf_token" value="{{ csrf_token() }}">{% endif %}
                                 <button type="submit" class="btn btn-danger">Delete</button>
                             </form>
@@ -149,4 +149,14 @@ <h2 class="section-title">Registered Users</h2>
         </table>
     </div>
 </section>
+<script>
+    document.querySelectorAll("form.js-confirm").forEach(function (form) {
+        form.addEventListener("submit", function (event) {
+            var message = form.getAttribute("data-confirm-message") || "Are you sure?";
+            if (!window.confirm(message)) {
+                event.preventDefault();
+            }
+        });
+    });
+</script>
 {% endblock %}

result_server/tests/test_admin_hardening.py

@@ -13,6 +13,8 @@
 
 install_portal_test_stubs()
 
+from utils.admin_policy import is_valid_email
+
 
 class _Store:
     def __init__(self):
@@ -89,6 +91,39 @@ def _cleanup(paths):
         shutil.rmtree(path)
 
 
+@pytest.mark.parametrize(
+    "email",
+    [
+        "user@example.com",
+        "first.last@example.co.jp",
+        "u+tag@example.com",
+        "user_name-123@a-b.example",
+    ],
+)
+def test_is_valid_email_accepts_well_formed_addresses(email):
+    assert is_valid_email(email)
+
+
+@pytest.mark.parametrize(
+    "email",
+    [
+        "evil';alert(1);//@x.com",
+        'user"@x.com',
+        "<script>@x.com",
+        "user @x.com",
+        "user\nname@x.com",
+        "",
+        "  ",
+        "no-at-sign",
+        "two@@at.com",
+        "user@.tld",
+        ("a" * 260) + "@x.com",
+    ],
+)
+def test_is_valid_email_rejects_dangerous_or_malformed_addresses(email):
+    assert not is_valid_email(email)
+
+
 def test_add_user_rejects_unknown_affiliation():
     store = _Store()
     store.create_user("admin@test.com", "SECRET", ["admin"])
@@ -110,6 +145,27 @@ def test_add_user_rejects_unknown_affiliation():
         _cleanup(temp_dirs)
 
 
+def test_add_user_rejects_xss_email_payload():
+    store = _Store()
+    store.create_user("admin@test.com", "SECRET", ["admin"])
+    app, temp_dirs = _admin_app(store)
+    try:
+        with app.test_client() as client:
+            _login_admin(client)
+            resp = client.post(
+                "/admin/users/add",
+                data={"email": "evil';alert(1);//@x.com", "affiliations": "dev"},
+                follow_redirects=True,
+            )
+
+        assert resp.status_code == 200
+        assert not store.user_exists("evil';alert(1);//@x.com")
+        assert store.invitations == []
+        assert b"Invalid email address" in resp.data
+    finally:
+        _cleanup(temp_dirs)
+
+
 def test_update_affiliations_rejects_unknown_affiliation():
     store = _Store()
     store.create_user("admin@test.com", "SECRET", ["admin"])
@@ -179,6 +235,25 @@ def test_add_user_accepts_arbitrary_safe_affiliation_without_allowlist():
         _cleanup(temp_dirs)
 
 
+def test_admin_users_template_does_not_embed_email_in_inline_submit_handler():
+    store = _Store()
+    store.create_user("admin@test.com", "SECRET", ["admin"])
+    store.create_user("evil';alert(1);x='@x.com", "SECRET2", ["dev"])
+    app, temp_dirs = _admin_app(store)
+    try:
+        with app.test_client() as client:
+            _login_admin(client)
+            resp = client.get("/admin/users")
+
+        html = resp.data.decode()
+        assert resp.status_code == 200
+        assert "onsubmit=" not in html
+        assert "data-confirm-message" in html
+        assert "evil';alert(1);x='" not in html
+    finally:
+        _cleanup(temp_dirs)
+
+
 def test_admin_email_routes_reject_path_segments():
     store = _Store()
     store.create_user("admin@test.com", "SECRET", ["admin"])

result_server/utils/admin_policy.py

@@ -2,9 +2,14 @@
 
 from __future__ import annotations
 
+import re
+
 _CONTROL_CHARS = frozenset(chr(code) for code in range(32)) | {chr(127)}
 _AFFILIATION_DELIMITERS = frozenset({","})
+_EMAIL_MAX_LEN = 254
+_EMAIL_RE = re.compile(r"^[A-Za-z0-9._%+\-]+@[A-Za-z0-9.\-]+\.[A-Za-z]{2,}$")
 _PATH_CHARS = frozenset({"/", "\\"})
+_UNSAFE_EMAIL_CHARS = frozenset({"<", ">", '"', "'", " ", "\t", "\n", "\r", "\\", "`"})
 
 
 def parse_allowed_affiliations(value: str | None) -> frozenset[str] | None:
@@ -16,6 +21,18 @@ def parse_allowed_affiliations(value: str | None) -> frozenset[str] | None:
     return frozenset(item.strip() for item in value.split(",") if item.strip())
 
 
+def is_valid_email(value: str) -> bool:
+    """Return whether a portal user email is safe and well formed."""
+    if not isinstance(value, str):
+        return False
+    email = value.strip()
+    if not (1 <= len(email) <= _EMAIL_MAX_LEN):
+        return False
+    if any(char in _UNSAFE_EMAIL_CHARS for char in email):
+        return False
+    return bool(_EMAIL_RE.fullmatch(email))
+
+
 def _is_safe_affiliation_name(value: str) -> bool:
     """Return whether a user-managed affiliation name is safe to store."""
     if not value or value in {".", ".."}: