diff --git a/user_scanner/user_scan/dev/pypi.py b/user_scanner/user_scan/dev/pypi.py index c1b5128f..6a2164e0 100644 --- a/user_scanner/user_scan/dev/pypi.py +++ b/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 `` 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)