Fix admin

app/admin/base.py

@@ -6,8 +6,11 @@
 from flask import redirect, url_for, request, flash, session, Response
 from flask_admin import expose, AdminIndexView, BaseView
 from flask_admin.contrib import sqla
+from flask_admin.contrib.sqla import tools
 from flask_login import current_user
 from markupsafe import Markup
+from sqlalchemy import Unicode, or_
+from sqlalchemy.sql.expression import cast
 from time import time
 
 from app import config
@@ -30,6 +33,24 @@ def _has_valid_admin_time() -> bool:
     return True
 
 
+_MAX_BIGINT = 2**63 - 1
+
+
+def _term_as_int(term: str) -> Optional[int]:
+    """Return the term as an int if it is a plain positive integer, else None.
+
+    flask-admin's `=` (exact match) prefix is accepted, as an exact match on a
+    number is exactly what the integer lookup does.
+    """
+    digits = term[1:] if term.startswith("=") else term
+    if not (digits.isascii() and digits.isdigit()):
+        return None
+    value = int(digits)
+    if value > _MAX_BIGINT:
+        return None
+    return value
+
+
 def _admin_action_formatter(view, context, model, name):
     action_name = AuditLogActionEnum.get_name(model.action)
     return "{} ({})".format(action_name, model.action)
@@ -81,6 +102,73 @@ class SLModelView(sqla.ModelView, BaseAdminView):
     can_delete = False
     edit_modal = True
 
+    def _apply_search(self, query, count_query, joins, count_joins, search):
+        """Numeric-aware version of flask-admin's search.
+
+        flask-admin casts *every* searchable column to text and matches it with
+        ILIKE '%term%'. On the user list that means searching an id runs
+        `CAST(users.id AS VARCHAR) ILIKE '%123%' OR CAST(users.email AS VARCHAR)
+        ILIKE '%123%'`, so both the list and the count query sequentially scan
+        the whole table.
+
+        Instead, when a term is a plain integer we compare the integer columns
+        directly (indexed lookup) and skip the text columns for that term; when
+        it is not, we skip the integer columns, which can never match by
+        equality. So a term is either an id lookup or a text search, never
+        both. To search text columns for a number, use flask-admin's `^`
+        (starts with) prefix.
+        """
+        for term in search.split(" "):
+            if not term:
+                continue
+
+            term_as_int = _term_as_int(term)
+            stmt = tools.parse_like_term(term)
+
+            int_filter, int_count_filter = [], []
+            text_filter, text_count_filter = [], []
+
+            for field, path in self._search_fields:
+                query, joins, alias = self._apply_path_joins(
+                    query, joins, path, inner_join=False
+                )
+
+                count_alias = None
+                if count_query is not None:
+                    count_query, count_joins, count_alias = self._apply_path_joins(
+                        count_query, count_joins, path, inner_join=False
+                    )
+
+                column = field if alias is None else getattr(alias, field.key)
+                count_column = (
+                    field if count_alias is None else getattr(count_alias, field.key)
+                )
+
+                # hybrid properties have no type, treat them as text
+                if isinstance(getattr(field, "type", None), sqlalchemy.Integer):
+                    if term_as_int is not None:
+                        int_filter.append(column == term_as_int)
+                        int_count_filter.append(count_column == term_as_int)
+                else:
+                    text_filter.append(cast(column, Unicode).ilike(stmt))
+                    text_count_filter.append(cast(count_column, Unicode).ilike(stmt))
+
+            # Prefer the integer columns, and only fall back to the text ones
+            # if the term produced no integer clause. If the term produced no
+            # clause at all - a non-numeric term on a view that only has
+            # integer searchable columns - nothing can match, and the filter
+            # must say so rather than be dropped.
+            filter_stmt = int_filter or text_filter or [sqlalchemy.false()]
+            count_filter_stmt = (
+                int_count_filter or text_count_filter or [sqlalchemy.false()]
+            )
+
+            query = query.filter(or_(*filter_stmt))
+            if count_query is not None:
+                count_query = count_query.filter(or_(*count_filter_stmt))
+
+        return query, count_query, joins, count_joins
+
     def on_model_change(self, form, model, is_created):
         changes = {}
         for attr in sqlalchemy.inspect(model).attrs:

app/admin/email_search.py

@@ -30,6 +30,7 @@
     Contact,
     EmailLog,
     Fido,
+    Subscription,
 )
 from app.alias_audit_log_utils import emit_alias_audit_log, AliasAuditLogAction
 from app.alias_delete import delete_alias as perform_alias_delete
@@ -331,6 +332,9 @@ class EmailSearchHelpers:
     PAGE_SIZE = 25
     ALIAS_DISPLAY_LIMIT = 5000
     UNPAGINATED_QUERY_LIMIT = 100
+    PADDLE_SUBSCRIPTION_URL = (
+        "https://vendors.paddle.com/subscriptions/customers/manage/{}"
+    )
 
     @staticmethod
     def mailbox_list(
@@ -439,6 +443,21 @@ def alias_total_pages(user: User) -> int:
     def partner_user(user: User) -> Optional[PartnerUser]:
         return PartnerUser.get_by(user_id=user.id)
 
+    @staticmethod
+    def paddle_subscription(user: User) -> Optional[Subscription]:
+        """Return the user's Paddle subscription, even if cancelled or expired.
+
+        Unlike User.get_paddle_subscription(), past subscriptions are returned
+        too, so admins can always reach the subscription in Paddle.
+        """
+        return Subscription.get_by(user_id=user.id)
+
+    @staticmethod
+    def paddle_subscription_url(subscription: Subscription) -> str:
+        return EmailSearchHelpers.PADDLE_SUBSCRIPTION_URL.format(
+            subscription.subscription_id
+        )
+
     @staticmethod
     def user_audit_log(user: User) -> list[UserAuditLog]:
         return (

templates/admin/email_search.html

@@ -28,6 +28,7 @@
   {% set pu = helper.partner_user(user) %}
   {% set active_sub = user.get_active_subscription() %}
   {% set sub_end = user.get_active_subscription_end() %}
+  {% set paddle_sub = helper.paddle_subscription(user) %}
   <div class="row mb-4 user-info-row">
     {# Account Information - 3/4 width #}
     <div class="col-lg-9 mb-3 mb-lg-0">
@@ -150,6 +151,30 @@ <h6 class="section-title">Subscription</h6>
                     </td>
                   </tr>
                 {% endif %}
+                {% if paddle_sub %}
+
+                  <tr>
+                    <td class="text-muted text-nowrap py-1">Paddle:</td>
+                    <td class="py-1">
+                      <a href="{{ helper.paddle_subscription_url(paddle_sub) }}"
+                         target="_blank"
+                         rel="noopener noreferrer">{{ paddle_sub.subscription_id }}</a>
+                      {{ copy_btn(paddle_sub.subscription_id) }}
+                      {% if paddle_sub.cancelled %}<span class="badge badge-danger">Cancelled</span>{% endif %}
+                      <br>
+                      <small class="text-muted">
+                        {{ paddle_sub.plan_name() }},
+                        {% if paddle_sub.cancelled or paddle_sub.next_bill_date < now.date() %}
+
+                          last bill
+                        {% else %}
+                          next bill
+                        {% endif %}
+                        {{ paddle_sub.next_bill_date }}
+                      </small>
+                    </td>
+                  </tr>
+                {% endif %}
                 <tr>
                   <td class="text-muted text-nowrap py-1">Paid:</td>
                   <td class="py-1">{{ badge_status(user.is_paid() , 'Yes', 'No', 'info', 'secondary') }}</td>

tests/admin/test_email_search.py

@@ -15,7 +15,10 @@
     Fido,
     AdminAuditLog,
     AuditLogActionEnum,
+    Subscription,
+    PlanEnum,
 )
+from app.admin.email_search import EmailSearchHelpers
 from app.proton.proton_partner import get_proton_partner
 from tests.utils import create_new_user, random_token
 
@@ -213,6 +216,82 @@ def test_email_search_user_with_subscription(flask_client):
     assert b"Subscription" in r.data  # Subscription section header
 
 
+def test_email_search_user_with_paddle_subscription_shows_link(flask_client):
+    """Test that a Paddle subscription links to the Paddle dashboard."""
+    login_admin(flask_client)
+
+    test_user = create_new_user(email=f"paddle_{random_token(8)}@example.com")
+    subscription_id = random_token(10)
+    Subscription.create(
+        user_id=test_user.id,
+        cancel_url="https://checkout.paddle.com/cancel",
+        update_url="https://checkout.paddle.com/update",
+        subscription_id=subscription_id,
+        event_time=arrow.now(),
+        next_bill_date=arrow.now().shift(months=1).date(),
+        plan=PlanEnum.monthly,
+        flush=True,
+    )
+    Session.commit()
+
+    r = flask_client.get(
+        url_for("admin.email_search.index"),
+        query_string={"query": test_user.email, "search_type": "email"},
+    )
+    assert r.status_code == 200
+    expected_url = EmailSearchHelpers.PADDLE_SUBSCRIPTION_URL.format(subscription_id)
+    assert expected_url.encode() in r.data
+    assert subscription_id.encode() in r.data
+
+
+def test_email_search_user_without_paddle_subscription_has_no_link(flask_client):
+    """Test that users without a Paddle subscription get no Paddle link."""
+    login_admin(flask_client)
+
+    test_user = create_new_user(email=f"nopaddle_{random_token(8)}@example.com")
+    Session.commit()
+
+    r = flask_client.get(
+        url_for("admin.email_search.index"),
+        query_string={"query": test_user.email, "search_type": "email"},
+    )
+    assert r.status_code == 200
+    assert b"vendors.paddle.com" not in r.data
+
+
+def test_email_search_expired_paddle_subscription_still_shows_link(flask_client):
+    """An expired Paddle subscription is still reachable from the admin panel."""
+    login_admin(flask_client)
+
+    test_user = create_new_user(email=f"oldpaddle_{random_token(8)}@example.com")
+    subscription_id = random_token(10)
+    Subscription.create(
+        user_id=test_user.id,
+        cancel_url="https://checkout.paddle.com/cancel",
+        update_url="https://checkout.paddle.com/update",
+        subscription_id=subscription_id,
+        event_time=arrow.now().shift(years=-1),
+        next_bill_date=arrow.now().shift(years=-1).date(),
+        plan=PlanEnum.yearly,
+        cancelled=True,
+        flush=True,
+    )
+    Session.commit()
+    # the subscription is past its billing date, so it is not the active one
+    assert test_user.get_paddle_subscription() is None
+
+    r = flask_client.get(
+        url_for("admin.email_search.index"),
+        query_string={"query": test_user.email, "search_type": "email"},
+    )
+    assert r.status_code == 200
+    expected_url = EmailSearchHelpers.PADDLE_SUBSCRIPTION_URL.format(subscription_id)
+    assert expected_url.encode() in r.data
+    assert b"Cancelled" in r.data
+    # a subscription that will never bill again isn't labelled "next bill"
+    assert b"last bill" in r.data
+
+
 def test_email_search_user_with_audit_logs(flask_client):
     """Test that user audit logs are displayed."""
     login_admin(flask_client)

tests/admin/test_model_view_search.py

@@ -0,0 +1,91 @@
+"""Tests for the numeric-aware search of the admin model views."""
+
+from app.admin.alias import AliasAdmin
+from app.admin.email_log import EmailLogAdmin
+from app.admin.user import UserAdmin
+from app.db import Session
+from app.models import Alias, EmailLog, User
+from tests.utils import create_new_user, random_token
+
+
+def _search_sql(view_cls, model, term: str) -> str:
+    view = view_cls(model, Session)
+    query, count_query, _, _ = view._apply_search(
+        Session.query(model), Session.query(model), {}, {}, term
+    )
+    return str(query)
+
+
+def test_numeric_search_looks_up_the_id_directly():
+    sql = _search_sql(UserAdmin, User, "12345")
+    assert "users.id = " in sql
+    # no full table scan on the text columns
+    assert "ILIKE" not in sql.upper()
+
+
+def test_non_numeric_search_only_hits_the_text_columns():
+    sql = _search_sql(UserAdmin, User, "bob@example.com")
+    assert "CAST(users.email AS VARCHAR) ILIKE" in sql
+    assert "users.id = " not in sql
+
+
+def test_numeric_search_on_joined_id_looks_up_the_id_directly():
+    sql = _search_sql(AliasAdmin, Alias, "12345")
+    assert "alias.id = " in sql
+    assert "ILIKE" not in sql.upper()
+
+
+def test_exact_match_prefix_still_looks_up_the_id_directly():
+    sql = _search_sql(UserAdmin, User, "=12345")
+    assert "users.id = " in sql
+    assert "ILIKE" not in sql.upper()
+
+
+def test_search_terms_that_are_not_plain_integers_keep_the_text_search():
+    # flask-admin's `^` (starts with) prefix, negative numbers and separators
+    # are not plain integers
+    for term in ("^12345", "-12345", "1_2345"):
+        sql = _search_sql(UserAdmin, User, term)
+        assert "CAST(users.email AS VARCHAR) ILIKE" in sql, term
+        assert "users.id = " not in sql, term
+
+
+def test_non_numeric_term_on_an_id_only_view_matches_nothing():
+    """A non-numeric term must not be silently dropped, which would list the
+    whole table. EmailLogAdmin only has an integer searchable column."""
+    sql = _search_sql(EmailLogAdmin, EmailLog, "foo@example.com")
+    assert "WHERE" in sql
+    view = EmailLogAdmin(EmailLog, Session)
+    query, count_query, _, _ = view._apply_search(
+        Session.query(EmailLog), Session.query(EmailLog), {}, {}, "foo@example.com"
+    )
+    assert query.all() == []
+    assert count_query.count() == 0
+
+
+def test_search_ignores_out_of_range_numbers():
+    sql = _search_sql(UserAdmin, User, str(2**64))
+    assert "users.id = " not in sql
+
+
+def test_numeric_search_finds_the_user(flask_client):
+    user = create_new_user(email=f"search_{random_token(8)}@example.com")
+    Session.commit()
+
+    view = UserAdmin(User, Session)
+    query, count_query, _, _ = view._apply_search(
+        Session.query(User), Session.query(User), {}, {}, str(user.id)
+    )
+    assert [found.id for found in query.all()] == [user.id]
+    assert count_query.count() == 1
+
+
+def test_numeric_search_does_not_match_ids_by_substring(flask_client):
+    user = create_new_user(email=f"substring_{random_token(8)}@example.com")
+    Session.commit()
+
+    view = UserAdmin(User, Session)
+    query, _, _, _ = view._apply_search(
+        Session.query(User), Session.query(User), {}, {}, str(user.id * 10)
+    )
+    assert user.id not in [found.id for found in query.all()]