Check email in breaches module (#56)

* Add unfinished email_breach module

* Add email breach module

* Remove debug code

* Skip tests on 429. Close session on error

* Make function for handling errors. Pretify get_random_string function.

* Fix response messages a little

Co-authored-by: manmolecular <regwebghost@yandex.ru>

Author: cravtos

src/scripts/osint/email_breach/__init__.py

@@ -0,0 +1,5 @@
+import sys
+from pathlib import Path
+
+__root_dir = Path(__file__).parents[4]
+sys.path.append(str(__root_dir))

src/scripts/osint/email_breach/__main__.py

@@ -0,0 +1,12 @@
+#!/usr/bin/env python3
+
+from pprint import pprint
+from sys import argv
+
+from src.core.utils.module import run_module
+from .module import Runner
+
+result = run_module(
+    Runner, args=argv, arg_name="email", arg_default="johndoe@gmail.com"
+)
+pprint(result)

src/scripts/osint/email_breach/module.py

@@ -0,0 +1,83 @@
+#!/usr/bin/env python3
+
+from hashlib import sha1
+from re import findall
+
+from bs4 import BeautifulSoup
+from requests import Session
+
+from src.core.base.osint import OsintRunner, PossibleKeys
+from src.core.utils.response import ScriptResponse
+from src.core.utils.validators import validate_kwargs
+
+
+class Defaults:
+    headers = {
+        "Content-Type": "application/x-www-form-urlencoded",
+        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) "
+        "Chrome/84.0.4147.105 Safari/537.36",
+    }
+
+
+class Runner(OsintRunner):
+    """
+    Check email in public breaches.
+    """
+
+    required = ["email"]
+
+    def __init__(self, logger: str = __name__):
+        super(Runner, self).__init__(logger)
+
+    @validate_kwargs(PossibleKeys.KEYS)
+    def run(self, *args, **kwargs) -> ScriptResponse.success or ScriptResponse.error:
+        """
+        Take email and look it up in breaches.
+        Breaches info is provided by monitor.firefox.com (haveibeenpwned.com)
+        """
+        email = kwargs.get("email")
+        if not isinstance(email, str):
+            return ScriptResponse.error(
+                result=None,
+                message=f"Can't make query. Incorrect input type (got {type(email)}, need {type('')}).",
+            )
+        email_hash = sha1(email.encode()).hexdigest()
+
+        with Session() as session:
+            session.headers.update(Defaults.headers)
+
+            resp = session.get(r"https://monitor.firefox.com")
+            if resp.status_code != 200:
+                return ScriptResponse.error(
+                    message=f"Can't look up email in breaches. Server response: {resp.status_code}.",
+                )
+
+            csrf_re = findall(r'(?<="_csrf" value=").*(?=">)', resp.text)
+            if not csrf_re:
+                return ScriptResponse.error(message=f"Can't find csrf token.")
+
+            csrf = csrf_re[0]
+            resp = session.post(
+                r"https://monitor.firefox.com/scan",
+                data={"_csrf": csrf, "emailHash": email_hash},
+            )
+            if resp.status_code != 200:
+                return ScriptResponse.error(
+                    message=f"Can't look up email in breaches. Server response: {resp.status_code}.",
+                )
+
+        breaches = []
+        soup = BeautifulSoup(resp.text, "html.parser")
+        for breach in soup.find_all("a", class_="breach-card"):
+            title = breach.find("span", class_="breach-title").text
+            info = breach.find_all("span", class_="breach-value")
+            if len(info) < 2:
+                continue
+            breaches.append(
+                {"title": title, "date": info[0].text, "compromised": info[1].text}
+            )
+
+        return ScriptResponse.success(
+            result=breaches,
+            message=f"Email {email} is found in {len(breaches)} breaches.",
+        )

src/scripts/osint/email_breach/requirements.txt

@@ -0,0 +1,12 @@
+beautifulsoup4==4.9.1
+certifi==2020.6.20
+chardet==3.0.4
+colorama==0.4.3
+commonmark==0.9.1
+idna==2.10
+Pygments==2.6.1
+requests==2.24.0
+rich==5.2.1
+soupsieve==2.0.1
+typing-extensions==3.7.4.2
+urllib3==1.25.10

src/scripts/osint/email_breach/test_module.py

@@ -0,0 +1,74 @@
+#!/usr/bin/env python3
+
+from unittest import TestCase, SkipTest
+from random import choices
+from string import ascii_uppercase
+
+from .module import Runner
+
+
+def get_random_string(length: int = 30) -> str:
+    """
+    Generates random string.
+    :param length: string length
+    :return: random string
+    """
+    return "".join(choices(ascii_uppercase, k=length))
+
+
+def check_response_msg(message: str = "") -> None:
+    """
+    Check that response message is valid and applicable
+    :param message: message of the response
+    :return: None
+    """
+    if "429" in message:
+        raise SkipTest("Server respond with 429 (Too many requests)")
+
+
+class EmailBreachTest(TestCase):
+    """
+    Defines basic tests for the email breach script.
+    """
+
+    def setUp(self):
+        """
+        Setup something before each test function
+        :return: None
+        """
+        self.runner = Runner()
+
+    def test_request(self) -> None:
+        """
+        Test email breach on johndoe@gmail.com.
+        :return: None
+        """
+        response = self.runner.run(email="johndoe@gmail.com")
+        check_response_msg(response.get("message", ""))
+
+        self.assertIn("found in", response.get("message"))
+        self.assertGreaterEqual(len(response.get("result")), 10)
+        self.assertIn(
+            {
+                "compromised": "Passwords, Email addresses",
+                "date": "December 4, 2013",
+                "title": "Adobe",
+            },
+            response.get("result"),
+        )
+
+    def test_random_string(self) -> None:
+        """
+        Test email breach on random string request.
+        """
+        response = self.runner.run(email=get_random_string())
+        check_response_msg(response.get("message", ""))
+        self.assertIn("found in", response.get("message"))
+
+    def test_unexpected_input(self) -> None:
+        """
+        Test email breach on unexpected input type
+        """
+        response = self.runner.run(email=None)
+        check_response_msg(response.get("message", ""))
+        self.assertIn("Can't make query", response.get("message"))