Merge pull request #174 from kaifcodec/add/email-osint-modules

add(pipeline/): new category pipeline with 4 new modules

Author: json-hunter07

user_scanner/email_scan/pipeline/__init__.py

@@ -0,0 +1,38 @@
+import httpx
+from user_scanner.core.result import Result
+
+
+async def _check(email: str) -> Result:
+    headers = {
+        'authority': 'axonaut.com',
+        '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,application/signed-exchange;v=b3;q=0.9',
+        'referer': 'https://axonaut.com/en',
+        'accept-language': 'en-US,en;q=0.9',
+    }
+
+    try:
+        async with httpx.AsyncClient(timeout=5.0, follow_redirects=False) as client:
+            response = await client.get(
+                f'https://axonaut.com/onboarding/?email={email}',
+                headers=headers
+            )
+
+            if response.status_code == 302 and "/login?email" in response.headers.get('Location', ''):
+
+                return Result.taken()
+
+            elif response.status_code == 200:
+                return Result.available()
+
+            else:
+                return Result.error(f"HTTP {response.status_code}")
+
+    except httpx.TimeoutException:
+        return Result.error("Connection timed out")
+    except Exception as e:
+        return Result.error(str(e))
+
+
+async def validate_axonaut(email: str) -> Result:
+    return await _check(email)

user_scanner/email_scan/pipeline/axonaut.py

@@ -0,0 +1,52 @@
+import httpx
+from user_scanner.core.result import Result
+
+
+async def _check(email: str) -> Result:
+    headers = {
+        'authority': 'api.hubspot.com',
+        'User-Agent': "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/144.0.0.0 Safari/537.36",
+        'content-type': 'application/json',
+        'origin': 'https://app.hubspot.com',
+        'referer': 'https://app.hubspot.com/',
+        'accept-language': 'en-US,en;q=0.9',
+    }
+
+    try:
+        async with httpx.AsyncClient(timeout=5.0, follow_redirects=True) as client:
+            payload = {
+                "email": email,
+                "password": "",
+                "rememberLogin": False
+            }
+
+            response = await client.post(
+                'https://api.hubspot.com/login-api/v1/login',
+                headers=headers,
+                json=payload
+            )
+
+            # HubSpot returns 400 for both "wrong password" and "user not found"
+            if response.status_code == 400:
+                data = response.json()
+                status = data.get("status")
+
+                if status == "INVALID_PASSWORD":
+                    return Result.taken()
+
+                elif status == "INVALID_USER":
+                    return Result.available()
+
+                else:
+                    return Result.error(data)
+
+            return Result.error(f"HTTP {response.status_code}")
+
+    except httpx.TimeoutException:
+        return Result.error("Connection timed out")
+    except Exception as e:
+        return Result.error(str(e))
+
+
+async def validate_hubspot(email: str) -> Result:
+    return await _check(email)

user_scanner/email_scan/pipeline/hubspot.py

@@ -0,0 +1,43 @@
+import httpx
+from user_scanner.core.result import Result
+
+
+async def _check(email: str) -> Result:
+    headers = {
+        'authority': 'accounts.insightly.com',
+        'accept': 'application/json, text/javascript, */*; q=0.01',
+        'x-requested-with': 'XMLHttpRequest',
+        'User-Agent': "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/144.0.0.0 Safari/537.36",
+        'content-type': 'application/x-www-form-urlencoded; charset=UTF-8',
+        'origin': 'https://accounts.insightly.com',
+        'referer': 'https://accounts.insightly.com/?plan=trial',
+        'accept-language': 'en-US,en;q=0.9',
+    }
+
+    try:
+        async with httpx.AsyncClient(timeout=5.0, follow_redirects=True) as client:
+            payload = {'emailaddress': email}
+
+            response = await client.post(
+                'https://accounts.insightly.com/signup/isemailvalid',
+                headers=headers,
+                data=payload
+            )
+
+            if "An account exists for this address." in response.text:
+                return Result.taken()
+
+            elif response.text.strip() == "true":
+                return Result.available()
+
+            else:
+                return Result.error(f"Unexpected response: {response.status_code}")
+
+    except httpx.TimeoutException:
+        return Result.error("Connection timed out")
+    except Exception as e:
+        return Result.error(str(e))
+
+
+async def validate_insightly(email: str) -> Result:
+    return await _check(email)

user_scanner/email_scan/pipeline/insightly.py

@@ -0,0 +1,63 @@
+import httpx
+from user_scanner.core.result import Result
+
+
+async def _check(email: str) -> Result:
+    headers = {
+        'User-Agent': "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/144.0.0.0 Safari/537.36",
+        'Content-Type': 'application/x-www-form-urlencoded;charset=UTF-8',
+        'Accept': '*/*',
+        'Origin': 'https://accounts.zoho.com',
+        'Sec-Fetch-Site': 'same-origin',
+        'Sec-Fetch-Mode': 'cors',
+        'Sec-Fetch-Dest': 'empty',
+        'Accept-Language': 'en-US,en;q=0.9',
+    }
+
+    try:
+        async with httpx.AsyncClient(timeout=5.0, follow_redirects=True) as client:
+            await client.get("https://accounts.zoho.com/register", headers=headers)
+
+            csrf_cookie = client.cookies.get("iamcsr")
+            if not csrf_cookie:
+                return Result.error("CSRF cookie not found")
+
+            headers['X-ZCSRF-TOKEN'] = f'iamcsrcoo={csrf_cookie}'
+
+            payload = {
+                'mode': 'primary',
+                'servicename': 'ZohoCRM',
+                'serviceurl': 'https://crm.zoho.com/crm/ShowHomePage.do',
+                'service_language': 'en'
+            }
+
+            response = await client.post(
+                f'https://accounts.zoho.com/signin/v2/lookup/{email}',
+                headers=headers,
+                data=payload
+            )
+
+            if response.status_code == 200:
+                data = response.json()
+                message = data.get("message")
+                status = data.get("status_code")
+
+                if message == "User exists" and status == 201:
+                    return Result.taken()
+
+                elif status == 400:
+                    return Result.available()
+
+                else:
+                    return Result.error(data)
+
+            return Result.error(f"HTTP {response.status_code}")
+
+    except httpx.TimeoutException:
+        return Result.error("Connection timed out")
+    except Exception as e:
+        return Result.error(str(e))
+
+
+async def validate_zoho(email: str) -> Result:
+    return await _check(email)