fix(pypi): stop reporting a package's publisher as the account holder

user_scanner/user_scan/dev/pypi.py

@@ -1,20 +1,30 @@
 import re
 import xmlrpc.client
+from email.utils import getaddresses
 from typing import Any
+
 import httpx
 
 from user_scanner.core.helpers import get_random_user_agent
 from user_scanner.core.orchestrator import Result, make_request
 
+# Packages listed on the result, and the ones sampled for release metadata.
+PACKAGE_SAMPLE = 5
+
+# Release metadata names whoever published a package — a co-maintainer, a
+# company, a mailing list — not necessarily the account it hangs off. The keys
+# below keep that provenance, so nothing downstream reads an address here as
+# the account holder's own mailbox or a name here as the account holder's name.
+CONTACT_ROLES = ("author", "maintainer")
+
 
 def validate_pypi(user: str) -> Result:
     """
     Validates a PyPI username and extracts:
 
-    - display_name
-    - email
     - packages_count
     - packages
+    - author / author_email / maintainer / maintainer_email of those packages
     """
 
     if not re.match(r"^(?!_+$)[A-Za-z0-9._-]+$", user):
@@ -26,13 +36,6 @@ def validate_pypi(user: str) -> Result:
     xmlrpc_url = "https://pypi.org/pypi"
     user_agent = get_random_user_agent()
 
-    extra: dict[str, Any] = {
-        "display_name": None,
-        "email": None,
-        "packages_count": 0,
-        "packages": [],
-    }
-
     #
     # XML-RPC lookup
     #
@@ -78,71 +81,70 @@ def validate_pypi(user: str) -> Result:
         return Result.available(url=profile_url)
 
     package_names = sorted({package_name for _, package_name in packages})
+    sample = package_names[:PACKAGE_SAMPLE]
 
-    extra["packages_count"] = len(package_names)
-    if len(package_names) > 5:
-        extra["packages"] = package_names[:5]
-    else:
-        extra["packages"] = package_names
-
-    #
-    # Query package JSON API to extract author name & email as fallback
-    #
-    if package_names:
-        name_candidate = None
-        email_candidate = None
-        for pkg in package_names:
-            if name_candidate and email_candidate:
-                break
-            try:
-                pkg_url = f"https://pypi.org/pypi/{pkg}/json"
-                res = make_request(
-                    pkg_url,
-                    headers={"User-Agent": user_agent},
-                    http2=True,
-                )
-                if res.status_code == 200:
-                    info = res.json().get("info", {})
-                    author = info.get("author")
-                    author_email = info.get("author_email")
-                    maintainer = info.get("maintainer")
-                    maintainer_email = info.get("maintainer_email")
-
-                    for email_field in (author_email, maintainer_email):
-                        if email_field and "@" in email_field:
-                            if "<" in email_field and ">" in email_field:
-                                parts = email_field.split("<")
-                                name_part = parts[0].strip()
-                                email_part = parts[1].replace(">", "").strip()
-                                if not name_candidate and name_part:
-                                    name_candidate = name_part
-                                if not email_candidate and email_part:
-                                    email_candidate = email_part
-                            elif not email_candidate:
-                                email_candidate = email_field.strip()
-
-                    # Fallbacks for names
-                    if (
-                        author
-                        and author.strip().lower() != "none"
-                        and not name_candidate
-                    ):
-                        name_candidate = author.strip()
-                    if (
-                        maintainer
-                        and maintainer.strip().lower() != "none"
-                        and not name_candidate
-                    ):
-                        name_candidate = maintainer.strip()
-            except Exception:
-                pass
-
-        if name_candidate:
-            extra["display_name"] = name_candidate
-        if email_candidate:
-            extra["email"] = email_candidate
+    extra: dict[str, Any] = {
+        "packages_count": len(package_names),
+        "packages": sample,
+    }
+    extra.update(_release_metadata(sample, user_agent))
 
     return Result.taken(
         url=profile_url,
         extra=extra,
     )
+
+
+def _release_metadata(packages: list[str], user_agent: str) -> dict[str, str]:
+    """Every distinct name and address the sampled packages credit.
+
+    All of them, rather than the first one an alphabetical walk reaches: which
+    co-publisher that lands on is an accident of package naming, and several
+    names on one account is the thing worth seeing.
+    """
+    found: dict[str, list[str]] = {}
+
+    for package in packages:
+        info = _package_info(package, user_agent)
+        for role in CONTACT_ROLES:
+            _add(found, role, info.get(role))
+            for name, address in _contacts(info.get(f"{role}_email")):
+                _add(found, role, name)
+                if "@" in address:
+                    _add(found, f"{role}_email", address)
+
+    return {key: ", ".join(values) for key, values in found.items()}
+
+
+def _package_info(package: str, user_agent: str) -> dict[str, Any]:
+    try:
+        response = make_request(
+            f"https://pypi.org/pypi/{package}/json",
+            headers={"User-Agent": user_agent},
+            http2=True,
+        )
+        if response.status_code != 200:
+            return {}
+        return response.json().get("info") or {}
+    except Exception:
+        return {}
+
+
+def _contacts(value: object) -> list[tuple[str, str]]:
+    """The ``Name <addr>`` pairs in a core metadata contact field, which may
+    credit several people in one comma-separated string."""
+    return [
+        (name.strip(), address.strip())
+        for name, address in getaddresses([str(value or "")])
+    ]
+
+
+def _add(found: dict[str, list[str]], key: str, value: object) -> None:
+    """Record a value under ``key``, unless it is empty, a duplicate, or the
+    literal ``None`` some releases ship where the field was left unset."""
+    text = str(value or "").strip()
+    if not text or text.lower() == "none":
+        return
+    values = found.setdefault(key, [])
+    if text not in values:
+        values.append(text)