-
Notifications
You must be signed in to change notification settings - Fork 1
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge branch 'main' of github.com:dmdhrumilmistry/breach-check
- Loading branch information
Showing
10 changed files
with
753 additions
and
451 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Empty file.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,32 @@ | ||
""" | ||
Base class for breach backends. | ||
""" | ||
from ..http import AsyncRequests | ||
|
||
|
||
class BaseBreachBackend: | ||
""" | ||
Base class for breach backends. | ||
Args: | ||
http_client (AsyncRequests): An instance of the HTTP client used for making requests. | ||
Attributes: | ||
_http_client (AsyncRequests): The HTTP client used for making requests. | ||
""" | ||
|
||
def __init__(self, http_client: AsyncRequests) -> None: | ||
self._http_client = http_client | ||
|
||
async def check_email_breaches(self, email: str): | ||
""" | ||
Check for breaches using the backend. | ||
Args: | ||
email (str): The email address to check for breaches. | ||
Returns: | ||
dict: A dictionary containing the results of the breach check. | ||
""" | ||
raise NotImplementedError | ||
|
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,57 @@ | ||
""" | ||
This module contains the implementation of MozillaMonitor class which checks email breaches using mozilla monitor API. | ||
""" | ||
from json import loads as json_loads | ||
from breach_check.breach_factory.base import BaseBreachBackend | ||
from breach_check.logger import logger | ||
from breach_check.http import AsyncRequests | ||
|
||
|
||
class LeakCheck(BaseBreachBackend): | ||
""" | ||
Checks email breaches using leakcheck public API. | ||
""" | ||
|
||
def __init__(self, http_client: AsyncRequests, *args, **kwargs) -> None: | ||
super().__init__(http_client, *args, **kwargs) | ||
self._api_url = 'https://leakcheck.io/api/public' | ||
|
||
async def check_email_breaches(self, email: str) -> dict: | ||
res_data = { | ||
'email': email, | ||
'breaches': [], | ||
'fields':[], | ||
'total': None | ||
} | ||
|
||
payload = { | ||
'check': email | ||
} | ||
response = await self._http_client.request( | ||
url=self._api_url, | ||
method='GET', | ||
params=payload | ||
) | ||
|
||
status_code = response.get('status') | ||
res_body = json_loads(response.get('res_body', '{}')) | ||
is_success = res_body.get('success', False) | ||
|
||
if status_code == 200 and is_success: | ||
logger.warning('Breaches found for %s', email) | ||
res_data['breaches'] = res_body.get('sources', []) | ||
res_data['fields'] = res_body.get('fields', []) | ||
res_data['total'] = res_body.get('found', -1) | ||
|
||
elif status_code == 200: | ||
logger.info('No breaches found for %s', email) | ||
res_data['total'] = 0 | ||
|
||
elif status_code == 429: | ||
logger.warning('Rate Limited') | ||
|
||
else: | ||
logger.error('Failed with status code: %s', str(status_code)) | ||
logger.error('Response: %s\nResponse Body: %s', str(response), str(res_body)) | ||
|
||
return res_data |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,54 @@ | ||
""" | ||
This module contains the implementation of MozillaMonitor class which checks email breaches using mozilla monitor API. | ||
""" | ||
from json import loads as json_loads | ||
from breach_check.breach_factory.base import BaseBreachBackend | ||
from breach_check.logger import logger | ||
from breach_check.http import AsyncRequests | ||
|
||
|
||
class MozillaMonitor(BaseBreachBackend): | ||
""" | ||
Checks email breaches using mozilla monitor API. | ||
""" | ||
|
||
def __init__(self, http_client: AsyncRequests, *args, **kwargs) -> None: | ||
logger.warning( | ||
'MozillaMonitor is deprecated since it API is no longer available. Please use other breach backends.') | ||
super().__init__(http_client, *args, **kwargs) | ||
self._api_url = 'https://monitor.firefox.com/api/v1/scan' | ||
|
||
async def check_email_breaches(self, email: str) -> dict: | ||
res_data = { | ||
'email': email, | ||
'breaches': [], | ||
'total': None | ||
} | ||
|
||
json_payload = { | ||
'email': email | ||
} | ||
response = await self._http_client.request( | ||
url=self._api_url, | ||
method='POST', | ||
json=json_payload, | ||
) | ||
|
||
status_code = response.get('status') | ||
res_body = json_loads(response.get('res_body', '{}')) | ||
is_success = res_body.get('success', False) | ||
|
||
if status_code == 200 and is_success: | ||
breaches = res_body.get('breaches', []) | ||
total = res_body.get('total', -1) | ||
res_data['breaches'] = breaches | ||
res_data['total'] = total | ||
|
||
elif status_code == 429: | ||
logger.warning('Rate Limited') | ||
|
||
else: | ||
logger.error('Failed with status code: %s', str(status_code)) | ||
logger.error(response, res_body) | ||
|
||
return res_data |
Oops, something went wrong.