Add LeakCheck service

README.md

@@ -73,7 +73,9 @@
 | [scylla.so](https://scylla.so/) - Service (free) | Cleartext passwords, hashs and salts, usernames, IPs, domain | :construction: |
 | [Dehashed.com](https://dehashed.com/) - Service | Cleartext passwords, hashs and salts, usernames, IPs, domain | :white_check_mark: :key: |
 | [IntelX.io](https://intelx.io/signup) - Service (free trial) | Cleartext passwords, hashs and salts, usernames, IPs, domain, Bitcoin Wallets, IBAN | :white_check_mark: :key: |
-| :new: [Breachdirectory.org](https://breachdirectory.org) - Service (free) | Cleartext passwords, hashs and salts, usernames, domain | :construction: :key: |
+| [Breachdirectory.org](https://breachdirectory.org) - Service (free) | Cleartext passwords, hashs and salts, usernames, domain | :construction: :key: |
+| :new: [LeakCheck.io](https://leakcheck.io/?from=h8mail) - Public | List of data breaches by e-mail or username and compromised fields | :white_check_mark: |
+| :new: [LeakCheck.io](https://leakcheck.io/?from=h8mail) - Service | Cleartext passwords, e-mails, usernames, ZIPs, IPs, full names, phones, stealer logs | :white_check_mark: :key: |
 
 *:key: - API key required*  
 

h8mail/utils/classes.py

@@ -9,6 +9,7 @@
 import platform
 from .version import __version__
 
+LC_SUPPORTED = ['email', 'domain', 'keyword', 'username', 'phone', 'phash', 'origin', 'password']
 
 class local_breach_target:
     """
@@ -750,6 +751,85 @@ def get_weleakinfo_pub(self, api_key):
             )
             print(ex)
 
+    def get_leakcheck_pub(self):
+        try:
+            c.info_news("[" + self.target + "]>[leakcheck public]")
+            url = "https://leakcheck.io/api/public?check={query}".format(
+                query=self.target
+            )
+            req = self.make_request(url, timeout=30)
+            response = req.json()
+            if req.status_code != 200:
+                c.bad_news(f"Got LC API response code {req.status_code} (public)")
+                return
+            else:
+                c.good_news(
+                    "Found {num} entries for {target} using LeakCheck (public)".format(
+                        num=response["found"], target=self.target
+                    )
+                )
+                if response["success"] is False:
+                    c.bad_news(response["error"])
+                    return
+                self.data.append(("LC_PUB_TOTAL", response["found"]))
+                if response["found"] == 0:
+                    return
+                for src in response["sources"]:
+                    self.data.append(("LC_PUB_SOURCE", src['name'] + " (" + src['date'] + ")"))
+                    self.pwned += 1
+        except Exception as ex:
+            c.bad_news(
+                f"LeakCheck error with {self.target} (public)"
+            )
+            print(ex)
+
+    def get_leakcheck_priv(self, api_key, user_query):
+        if user_query == 'hash': user_query = 'phash'
+
+        if user_query not in LC_SUPPORTED:
+            c.bad_news(
+                f"LeakCheck does not support {user_query} search (yet)"
+            )
+            return
+        try:
+            c.info_news("[" + self.target + "]>[leakcheck pro]")
+            url = "https://leakcheck.io/api/v2/query/{query}?type={user_query}".format(
+                query=self.target,
+                user_query=user_query
+            )
+            self.headers.update({"X-API-KEY": api_key})
+            req = self.make_request(url, timeout=30)
+            response = req.json()
+            if req.status_code != 200:
+                c.bad_news(f"Got LC API response code {req.status_code} - {response.get('error')} (Pro)")
+                return
+            else:
+                c.good_news(
+                    "Found {num} entries for {target} using LeakCheck (Pro)".format(
+                        num=response["found"], target=self.target
+                    )
+                )
+                self.data.append(("LC_PRIV_TOTAL", response["found"]))
+                if response["found"] == 0:
+                    return
+                for result in response["result"]:
+                    breach_name = result['source']['name'] or "N/A"
+                    breach_date = result['source']['breach_date'] or "Unknown"
+                    self.data.append(("LC_PRIV_SOURCE", breach_name + " (" + breach_date + ")"))
+                    self.pwned += 1
+                    result['fields'].remove(user_query)
+                    for field in result['fields']:
+                        field_name = field
+                        if field == 'username': field_name = 'USER'
+                        if field == 'password': field_name = 'PASS'
+
+                        self.data.append(("LC_PRIV_" + field_name.upper(), result[field]))
+        except Exception as ex:
+            c.bad_news(
+                f"LeakCheck error with {self.target} (private)"
+            )
+            print(ex)
+
     def get_dehashed(self, api_email, api_key, user_query):
         try:
             if user_query == "hash":

h8mail/utils/gen_config.py

@@ -24,6 +24,7 @@ def gen_config_file():
 ;intelx_maxfile = 10
 ;breachdirectory_user = 
 ;breachdirectory_pass =
+;leakcheck_apikey =
 """
         dest_config.write(config)
         c.good_news(

h8mail/utils/print_results.py

@@ -49,6 +49,8 @@ def print_results(results, hide=False):
                     c.print_result(t.target, t.data[i][1], t.data[i][0])
                 if "WLI" in t.data[i][0]:
                     c.print_result(t.target, t.data[i][1], t.data[i][0])
+                if "LC" in t.data[i][0]:
+                    c.print_result(t.target, t.data[i][1], t.data[i][0])
                 if "SCYLLA" in t.data[i][0]:
                     c.print_result(t.target, t.data[i][1], t.data[i][0])
                 if "DHASHD" in t.data[i][0]:

h8mail/utils/run.py

@@ -8,7 +8,7 @@
 import sys
 
 from .breachcompilation import breachcomp_check
-from .classes import target
+from .classes import target, LC_SUPPORTED
 from .colors import colors as c
 from .helpers import (
     fetch_emails,
@@ -64,6 +64,7 @@ def target_factory(targets, user_args):
             current_target = target(t)
         if not skip_default_queries:
             if not user_args.skip_defaults:
+                current_target.get_leakcheck_pub()
                 current_target.get_hunterio_public()
                 ## emailrep seems to insta-block h8mail user agent without a key
                 # if api_keys is None or "emailrep" not in api_keys:
@@ -83,6 +84,8 @@ def target_factory(targets, user_args):
                 )
             if "hibp" in api_keys and query == "email":
                 current_target.get_hibp3(api_keys["hibp"])
+            if "leakcheck_apikey" in api_keys and query in LC_SUPPORTED:
+                current_target.get_leakcheck_priv(api_keys["leakcheck_apikey"], query)
             if "emailrep" in api_keys and query == "email":
                 current_target.get_emailrepio(api_keys["emailrep"])
             if "hunterio" in api_keys and query == "email":
@@ -284,7 +287,7 @@ def parse_args(args):
         "-sk",
         "--skip-defaults",
         dest="skip_defaults",
-        help="Skips Scylla and HunterIO check. Ideal for local scans",
+        help="Skips Scylla, LeakCheck and HunterIO check. Ideal for local scans",
         action="store_true",
         default=False,
     )