fix: improve Gumroad email and username detection (#627)

- Use signup validation for email checks
- Parse profile metadata and verify username matches

Author: kristoisberg

user_scanner/email_scan/creator/gumroad.py

@@ -1,83 +1,64 @@
-import httpx
 import re
-from user_scanner.core.result import Result
-
 
-async def _check(email: str) -> Result:
-    show_url = "https://gumroad.com"
-    async with httpx.AsyncClient(timeout=15.0, http2=False, follow_redirects=True) as client:
-        try:
-            url1 = "https://gumroad.com/users/forgot_password/new"
-            headers1 = {
-                'User-Agent': "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/144.0.0.0 Safari/537.36",
-                'Accept': "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8",
-                'Accept-Encoding': "identity",
-                'sec-ch-ua': '"Not(A:Brand";v="8", "Chromium";v="144", "Google Chrome";v="144"',
-                'sec-ch-ua-mobile': "?0",
-                'sec-ch-ua-platform': '"Linux"',
-                'upgrade-insecure-requests': "1",
-                'referer': "https://www.google.com/",
-                'accept-language': "en-US,en;q=0.9"
-            }
+from user_scanner.core.impersonate import impersonate_request_async
+from user_scanner.core.result import Result
 
-            res1 = await client.get(url1, headers=headers1)
-            html = res1.text
+SIGNUP_URL = "https://gumroad.com/signup"
+CSRF_RE = re.compile(r'<meta name="csrf-token" content="([^"]+)"')
 
-            csrf_match = re.search(
-                r'authenticity_token&quot;:&quot;([^&]+)&quot;', html)
-            if not csrf_match:
-                csrf_match = re.search(
-                    r'name="csrf-token" content="([^"]+)"', html)
 
-            if not csrf_match:
-                return Result.error("Failed to extract CSRF token")
+async def validate_gumroad(email: str) -> Result:
+    show_url = "https://gumroad.com"
 
-            csrf_token = csrf_match.group(1)
+    try:
+        page = await impersonate_request_async(SIGNUP_URL, allow_redirects=True)
+        if page.status_code != 200:
+            return Result.error(
+                f"Unexpected Gumroad signup response: {page.status_code}",
+                url=show_url,
+            )
 
-            url2 = "https://gumroad.com/users/forgot_password"
+        token = CSRF_RE.search(page.text)
+        if not token or "Signup/New" not in page.text:
+            return Result.error("Could not read Gumroad signup form", url=show_url)
 
-            payload = {
+        response = await impersonate_request_async(
+            SIGNUP_URL,
+            "POST",
+            json={
                 "user": {
-                    "email": email
-                }
-            }
+                    "email": email,
+                    # Gumroad rejects this three-character password after checking
+                    # whether the email is already registered.
+                    "password": "Q7~",
+                },
+            },
+            headers={
+                "accept": "text/html, application/xhtml+xml",
+                "origin": "https://gumroad.com",
+                "referer": SIGNUP_URL,
+                "x-csrf-token": token.group(1),
+                "x-inertia": "true",
+                "x-requested-with": "XMLHttpRequest",
+            },
+            allow_redirects=True,
+        )
+        if response.status_code != 200:
+            return Result.error(
+                f"Unexpected Gumroad signup response: {response.status_code}",
+                url=show_url,
+            )
 
-            headers2 = {
-                'User-Agent': "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/144.0.0.0 Safari/537.36",
-                'Accept': "text/html, application/xhtml+xml",
-                'Accept-Encoding': "identity",
-                'Content-Type': "application/json",
-                'sec-ch-ua-platform': '"Linux"',
-                'x-csrf-token': csrf_token,
-                'sec-ch-ua': '"Not(A:Brand";v="8", "Chromium";v="144", "Google Chrome";v="144"',
-                'x-inertia': "true",
-                'sec-ch-ua-mobile': "?0",
-                'x-requested-with': "XMLHttpRequest",
-                'origin': "https://gumroad.com",
-                'sec-fetch-site': "same-origin",
-                'sec-fetch-mode': "cors",
-                'sec-fetch-dest': "empty",
-                'referer': "https://gumroad.com/users/forgot_password/new",
-                'accept-language': "en-US,en;q=0.9",
-                'priority': "u=1, i"
-            }
+        data = response.json()
+        if data.get("component") != "Signup/New":
+            return Result.error("Unexpected Gumroad signup page", url=show_url)
 
-            response = await client.post(url2, json=payload, headers=headers2)
-
-            data = response.json()
-            flash_msg = data.get("props", {}).get(
-                "flash", {}).get("message", "")
-
-            if "An account does not exist" in flash_msg:
-                return Result.available(url=show_url)
-            elif "An account does not exist" not in flash_msg:
-                return Result.taken(url=show_url)
-            else:
-                return Result.error(f"Unexpected status: {response.status_code}")
-
-        except Exception as e:
-            return Result.error(f"unexpected exception: {e}")
-
-
-async def validate_gumroad(email: str) -> Result:
-    return await _check(email)
+        flash = data.get("props", {}).get("flash") or {}
+        message = flash.get("message")
+        if message == "An account already exists with this email.":
+            return Result.taken(url=show_url)
+        if message == "Password is too short (minimum is 4 characters)":
+            return Result.available(url=show_url)
+        return Result.error("Unexpected Gumroad signup result", url=show_url)
+    except Exception as exc:
+        return Result.error(exc, url=show_url)

user_scanner/user_scan/creator/gumroad.py

@@ -1,28 +1,91 @@
+import html
+import json
 import re
-from user_scanner.core.orchestrator import Result, make_request
+from typing import cast
+
+from user_scanner.core.orchestrator import generic_validate
+from user_scanner.core.result import Result
+
+NOT_FOUND = "<title>Page not found (404) - Gumroad</title>"
+
+
+def _links(value):
+    if isinstance(value, dict):
+        if isinstance(value.get("href"), str):
+            yield value["href"]
+        for item in value.values():
+            yield from _links(item)
+    elif isinstance(value, list):
+        for item in value:
+            yield from _links(item)
 
 
 def validate_gumroad(user: str) -> Result:
-    if not re.fullmatch(r"[a-z0-9]{3,20}", user):
+    username = user.lower()
+    if not re.fullmatch(r"(?=.*[a-z])[a-z0-9]{3,20}", username):
         return Result.error(
-            "Username must be between 3 and 20 lowercase alphanumeric characters"
+            "Username must be 3-20 lowercase letters and numbers, with at least one letter",
+            url="https://gumroad.com",
+        )
+
+    url = f"https://{username}.gumroad.com/"
+
+    def process(response) -> Result:
+        if response.status_code == 404 and NOT_FOUND in response.text:
+            return Result.available()
+
+        match = re.search(r'<div id="app" data-page="([^"]+)"', response.text)
+        if response.status_code != 200 or not match:
+            return Result.error(f"Unexpected Gumroad response: {response.status_code}")
+
+        try:
+            page = json.loads(html.unescape(match.group(1)))
+        except json.JSONDecodeError:
+            return Result.error("Invalid Gumroad profile data")
+
+        props = page.get("props", {})
+        profile = props.get("creator_profile")
+        if (
+            not isinstance(profile, dict)
+            or profile.get("subdomain") != f"{username}.gumroad.com"
+        ):
+            return Result.error("Gumroad profile did not match the requested username")
+
+        reputation = profile.get("reputation") or {}
+        preview = next(
+            (
+                tag.get("content")
+                for tag in props.get("_inertia_meta", [])
+                if tag.get("property") == "og:image"
+            ),
+            None,
+        )
+        links = list(
+            dict.fromkeys(
+                _links(
+                    [
+                        section.get("text")
+                        for section in props.get("sections", [])
+                        if section.get("type") == "SellerProfileRichTextSection"
+                    ]
+                )
+            )
+        )
+        result = Result.taken(
+            extra={
+                "uid": profile.get("external_id"),
+                "name": profile.get("name"),
+                "bio": props.get("bio"),
+                "twitter_handle": profile.get("twitter_handle"),
+                "verified": profile.get("is_verified"),
+                "rating": reputation.get("average"),
+                "reviews": reputation.get("count"),
+                "products": reputation.get("products_count"),
+            },
+            media={"avatar": profile.get("avatar_url"), "preview": preview},
         )
+        if links:
+            cast(dict, result.extra)["links"] = links
+        return result
 
-    url = f"https://{user}.gumroad.com/"
-    show_url = f"https://{user}.gumroad.com"
-    
-    try:
-        response = make_request(url, follow_redirects=True)
-        if response.status_code == 200:
-            html = response.text
-            extra = {}
-            title = re.search(r'<title>([^\|]+)\|', html)
-            if title:
-                extra["name"] = title.group(1).strip()
-            return Result.taken(extra=extra, url=show_url)
-        elif response.status_code == 404:
-            return Result.available(url=show_url)
-        else:
-            return Result.error(f"Unexpected status: {response.status_code}", url=show_url)
-    except Exception as e:
-        return Result.error(e, url=show_url)
+    return generic_validate(url, process, show_url=url, follow_redirects=True)