feat(admin): manually execute domain check

Signed-off-by: Mike Fiedler <[email protected]>

Author: miketheman

tests/unit/accounts/test_core.py

@@ -17,6 +17,7 @@
 
 from warehouse import accounts
 from warehouse.accounts.interfaces import (
+    IDomainStatusService,
     IEmailBreachedService,
     IPasswordBreachedService,
     ITokenService,
@@ -25,6 +26,7 @@
 from warehouse.accounts.services import (
     HaveIBeenPwnedEmailBreachedService,
     HaveIBeenPwnedPasswordBreachedService,
+    NullDomainStatusService,
     TokenServiceFactory,
     database_login_factory,
 )
@@ -186,6 +188,7 @@ def test_includeme(monkeypatch):
             HaveIBeenPwnedEmailBreachedService.create_service,
             IEmailBreachedService,
         ),
+        pretend.call(NullDomainStatusService.create_service, IDomainStatusService),
         pretend.call(RateLimit("10 per 5 minutes"), IRateLimiter, name="user.login"),
         pretend.call(RateLimit("10 per 5 minutes"), IRateLimiter, name="ip.login"),
         pretend.call(

tests/unit/admin/test_routes.py

@@ -89,6 +89,13 @@ def test_includeme():
             factory="warehouse.accounts.models:UserFactory",
             traverse="/{username}",
         ),
+        pretend.call(
+            "admin.user.email_domain_check",
+            "/admin/users/{username}/email_domain_check/",
+            domain=warehouse,
+            factory="warehouse.accounts.models:UserFactory",
+            traverse="/{username}",
+        ),
         pretend.call(
             "admin.user.delete",
             "/admin/users/{username}/delete/",

warehouse/accounts/__init__.py

@@ -13,6 +13,7 @@
 from celery.schedules import crontab
 
 from warehouse.accounts.interfaces import (
+    IDomainStatusService,
     IEmailBreachedService,
     IPasswordBreachedService,
     ITokenService,
@@ -25,6 +26,7 @@
 from warehouse.accounts.services import (
     HaveIBeenPwnedEmailBreachedService,
     HaveIBeenPwnedPasswordBreachedService,
+    NullDomainStatusService,
     NullEmailBreachedService,
     NullPasswordBreachedService,
     TokenServiceFactory,
@@ -131,6 +133,14 @@ def includeme(config):
         breached_email_class.create_service, IEmailBreachedService
     )
 
+    # Register our domain status service.
+    domain_status_class = config.maybe_dotted(
+        config.registry.settings.get("domain_status.backend", NullDomainStatusService)
+    )
+    config.register_service_factory(
+        domain_status_class.create_service, IDomainStatusService
+    )
+
     # Register our security policies.
     config.set_security_policy(
         MultiSecurityPolicy(

warehouse/accounts/interfaces.py

@@ -298,3 +298,10 @@ def get_email_breach_count(email: str) -> int | None:
         """
         Returns count of times the email appears in verified breaches.
         """
+
+
+class IDomainStatusService(Interface):
+    def get_domain_status(domain: str) -> list[str]:
+        """
+        Returns a list of status strings for the given domain.
+        """

warehouse/accounts/services.py

@@ -10,6 +10,8 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+from __future__ import annotations
+
 import collections
 import datetime
 import functools
@@ -18,6 +20,7 @@
 import logging
 import os
 import secrets
+import typing
 import urllib.parse
 
 import passlib.exc
@@ -35,6 +38,7 @@
 
 from warehouse.accounts.interfaces import (
     BurnedRecoveryCode,
+    IDomainStatusService,
     IEmailBreachedService,
     InvalidRecoveryCode,
     IPasswordBreachedService,
@@ -62,6 +66,9 @@
 from warehouse.rate_limiting import DummyRateLimiter, IRateLimiter
 from warehouse.utils.crypto import BadData, SignatureExpired, URLSafeTimedSerializer
 
+if typing.TYPE_CHECKING:
+    from pyramid.request import Request
+
 logger = logging.getLogger(__name__)
 
 PASSWORD_FIELD = "password"
@@ -962,3 +969,47 @@ def create_service(cls, context, request):
     def get_email_breach_count(self, email):
         # This service allows *every* email as a non-breached email.
         return 0
+
+
+@implementer(IDomainStatusService)
+class NullDomainStatusService:
+    @classmethod
+    def create_service(cls, _context, _request):
+        return cls()
+
+    def get_domain_status(self, _domain: str) -> list[str]:
+        return ["active"]
+
+
+@implementer(IDomainStatusService)
+class DomainrDomainStatusService:
+    def __init__(self, session, api_key):
+        self._http = session
+        self.api_key = api_key
+
+    @classmethod
+    def create_service(cls, _context, request: Request) -> DomainrDomainStatusService:
+        domainr_api_key = request.registry.settings.get("domainr.api_key")
+        return cls(session=request.http, api_key=domainr_api_key)
+
+    def get_domain_status(self, domain: str) -> list[str]:
+        """
+        Check if a domain is available or not.
+        See https://domainr.com/docs/api/v2/status
+        """
+        # bail early if no api key is set, so we don't send failing requests
+        if not self.api_key:
+            return []
+
+        try:
+            resp = self._http.get(
+                "https://api.domainr.com/v2/status",
+                params={"client_id": self.api_key, "domain": domain},
+                timeout=5,
+            )
+            resp.raise_for_status()
+        except requests.RequestException as exc:
+            logger.warning("Error contacting Domainr: %r", exc)
+            return []
+
+        return resp.json()["status"][0]["status"].split()

warehouse/admin/routes.py

@@ -87,6 +87,13 @@ def includeme(config):
         factory="warehouse.accounts.models:UserFactory",
         traverse="/{username}",
     )
+    config.add_route(
+        "admin.user.email_domain_check",
+        "/admin/users/{username}/email_domain_check/",
+        factory="warehouse.accounts.models:UserFactory",
+        domain=warehouse,
+        traverse="/{username}",
+    )
     config.add_route(
         "admin.user.delete",
         "/admin/users/{username}/delete/",

warehouse/admin/templates/admin/users/detail.html

@@ -522,49 +522,45 @@ <h3 class="card-title">Emails</h3>
                     <div class="alert alert-danger"><i class="icon fa fa-ban"></i> {{ error }}</div>
                   {% endfor %}
                 {% endif %}
-
-                {% for field in emails_form.emails.entries %}
-                <div class="row">
-                  <div class="col-sm-7">
-                    {{ render_field("Email", field.email, "email-" ~ loop.index0, class="form-control", placeholder="Email", permission=perms_admin_users_email_write)}}
-                  </div>
-                  <div class="col-sm">
-                    {{ render_checkbox("Primary", field.primary, "email-primary-" ~ loop.index0, permission=perms_admin_users_email_write)}}
-                  </div>
-                  <div class="col-sm">
-                    {{ render_checkbox("Verified", field.verified, "email-verified-" ~ loop.index0, permission=perms_admin_users_email_write) }}
-                  </div>
-                  <div class="col-sm">
-                    {{ render_checkbox("Public", field.public, "email-public-" ~ loop.index0, permission=perms_admin_users_email_write) }}
-                  </div>
-                  {% if breached_email_count[field.email.data] %}
-                  <div class="col-sm">
-                    Breaches: {{ breached_email_count[field.email.data] }}
-                  </div>
-                  {% endif %}
-                  {% if field.unverify_reason.data %}
-                  <div class="col">
-                    <i class="fa fa-times text-red"></i> Unverify Reason: {{ field.unverify_reason.data.value }}
-                  </div>
-                  {% endif %}
-                </div>
                 <div class="row">
                   <div class="col">
-                    Domain Status: {{ field.domain_last_status.data }}
-                  </div>
-                  <div class="col">
-                    Last checked: {{ field.domain_last_checked.data }}
+                    {% for field in emails_form.emails.entries %}
+                      <div class="row">
+                        <div class="col-sm-6">
+                          {{ render_field("Email", field.email, "email-" ~ loop.index0, class="form-control", placeholder="Email", permission=perms_admin_users_email_write) }}
+                        </div>
+                        <div class="col-sm">
+                          {{ render_checkbox("Primary", field.primary, "email-primary-" ~ loop.index0, permission=perms_admin_users_email_write) }}
+                        </div>
+                        <div class="col-sm">
+                          {{ render_checkbox("Verified", field.verified, "email-verified-" ~ loop.index0, permission=perms_admin_users_email_write) }}
+                        </div>
+                        <div class="col-sm">
+                          {{ render_checkbox("Public", field.public, "email-public-" ~ loop.index0, permission=perms_admin_users_email_write) }}
+                        </div>
+                        {% if breached_email_count[field.email.data] %}
+                          <div class="col-sm">Breaches: {{ breached_email_count[field.email.data] }}</div>
+                        {% endif %}
+                        {% if field.unverify_reason.data %}
+                          <div class="col">
+                            <i class="fa fa-times text-red"></i> Unverify Reason: {{ field.unverify_reason.data.value }}
+                          </div>
+                        {% endif %}
+                      </div>
+                      <div class="row">
+                        <div class="col-sm">Domain Status: {{ field.domain_last_status.data }}</div>
+                        <div class="col-sm">Last checked: {{ field.domain_last_checked.data }}</div>
+                        <div class="col-sm">
+                          <button type="button"
+                                  class="btn btn-sm btn-info"
+                                  data-toggle="modal"
+                                  data-target="#checkEmailDomain-{{ field.email.id }}">Check now</button>
+                        </div>
+                      </div>
+                      {% if not loop.last %}<hr />{% endif %}
+                    {% endfor %}
                   </div>
-                  {# TODO: Uncomment after implementing an admin route to run the check and flash, but needs to work for each email #}
-                  {# <div class="col">#}
-                  {# <button type="button" class="btn btn-primary" data-toggle="modal" data-target="#checkDomainModal">#}
-                  {# <i class="fa-solid fa-magnifying-glass"></i>#}
-                  {# Check Domain Now #}
-                  {# </button>#}
-                  {# </div>#}
                 </div>
-                {% if not loop.last %}<hr />{% endif %}
-                {% endfor %}
               </div>
 
               <div class="card-footer">
@@ -1027,4 +1023,37 @@ <h3 class="card-title">Journals (most recent)</h3>
 
     </div>
   </div>
+  {% for email in emails_form.emails.entries %}
+    <div class="modal fade"
+         id="checkEmailDomain-{{ email.id }}-email"
+         tabindex="-1"
+         role="dialog">
+      <form method="POST"
+            action="{{ request.route_path('admin.user.email_domain_check', username=user.username, email_address=email.email) }}">
+        <input name="csrf_token"
+               type="hidden"
+               value="{{ request.session.get_csrf_token() }}">
+        <input name="email_address" type="hidden" value="{{ email.email.data }}">
+        <div class="modal-dialog" role="document">
+          <div class="modal-content">
+            <div class="modal-header">
+              <h4 class="modal-title" id="checkDomainStatusModalLabel">Check domain status?</h4>
+              <button type="button" class="close" data-dismiss="modal">
+                <span>&times;</span>
+              </button>
+            </div>
+            <div class="modal-body">
+              <p>Perform a domain status lookup check and update the database with the results.</p>
+              <p>Check domain for email address:</p>
+              <code>{{ email.email.data }}</code>
+            </div>
+            <div class="modal-footer">
+              <button type="button" class="btn btn-secondary" data-dismiss="modal">Close</button>
+              <button type="submit" class="btn btn-info">Check domain</button>
+            </div>
+          </div>
+        </div>
+      </form>
+    </div>
+  {% endfor %}
 {% endblock %}

warehouse/admin/views/users.py

@@ -28,6 +28,7 @@
 
 from warehouse.accounts.interfaces import (
     BurnedRecoveryCode,
+    IDomainStatusService,
     IEmailBreachedService,
     InvalidRecoveryCode,
     IUserService,
@@ -665,3 +666,33 @@ def user_burn_recovery_codes(user, request):
 
         request.session.flash(f"Burned {n_burned} recovery code(s)", queue="success")
     return HTTPSeeOther(request.route_path("admin.user.detail", username=user.username))
+
+
+@view_config(
+    route_name="admin.user.email_domain_check",
+    require_methods=["POST"],
+    permission=Permissions.AdminUsersEmailWrite,
+    uses_session=True,
+    require_csrf=True,
+    context=User,
+)
+def user_email_domain_check(user, request):
+    """
+    Run a background check on the email domain of the user.
+    Flash the user that the check has been done.
+    """
+    email_address = request.params.get("email_address")
+    email = request.db.scalar(select(Email).where(Email.email == email_address))
+
+    domain_status_service = request.find_service(IDomainStatusService)
+    domain_status = domain_status_service.get_domain_status(email_address)
+
+    # set the domain status to the email address
+    email.domain_last_checked = datetime.datetime.now(datetime.UTC)
+    email.domain_last_status = domain_status
+    request.db.add(email)
+
+    request.session.flash(
+        f"Domain status check for {email_address!r} completed", queue="success"
+    )
+    return HTTPSeeOther(request.route_path("admin.user.detail", username=user.username))