diff --git a/artemis/config.py b/artemis/config.py index 5ec800c81..20c7f1c45 100644 --- a/artemis/config.py +++ b/artemis/config.py @@ -437,6 +437,11 @@ class WordPressBruter: "www123, when testing www.projectname.example.com.", ] = get_config("WORDPRESS_BRUTER_STRIPPED_PREFIXES", default="www", cast=decouple.Csv(str)) + class DomainExpirationScanner: + DOMAIN_EXPIRATION_TIMEFRAME_DAYS: Annotated[ + int, "The scanner warns if the domain's expiration date falls within this time frame from now." + ] = get_config("DOMAIN_EXPIRATION_TIMEFRAME_DAYS", default=5, cast=int) + @staticmethod def verify_each_variable_is_annotated() -> None: def verify_class(cls: type) -> None: diff --git a/artemis/modules/domain_expiration_scanner.py b/artemis/modules/domain_expiration_scanner.py new file mode 100644 index 000000000..08363ab9c --- /dev/null +++ b/artemis/modules/domain_expiration_scanner.py @@ -0,0 +1,74 @@ +#!/usr/bin/env python3 +import datetime +import time +from typing import Any, Dict, Optional + +from karton.core import Task +from whois import Domain, WhoisQuotaExceeded, query # type: ignore + +from artemis.binds import TaskStatus, TaskType +from artemis.config import Config +from artemis.domains import is_main_domain +from artemis.module_base import ArtemisBase + + +class DomainExpirationScanner(ArtemisBase): + """ + Alerts if domain expiration date is coming. + """ + + identity = "domain_expiration_scanner" + filters = [{"type": TaskType.DOMAIN.value}] + + def run(self, current_task: Task) -> None: + domain = current_task.get_payload(TaskType.DOMAIN) + result: Dict[str, Any] = {} + status = TaskStatus.OK + status_reason = None + if is_main_domain(domain): + try: + domain_data = self._query_whois(domain=domain) + except WhoisQuotaExceeded: + time.sleep(24 * 60 * 60) + domain_data = self._query_whois(domain=domain) + + expiry_date = domain_data.expiration_date + result = self._prepare_expiration_data(expiration_date=expiry_date, result=result) + + if "close_expiration_date" in result: + status = TaskStatus.INTERESTING + status_reason = self._prepare_expiration_status_reason( + days_to_expire=result["days_to_expire"], expiration_date=result["expiration_date"] + ) + + self.db.save_task_result(task=current_task, status=status, status_reason=status_reason, data=result) + + @staticmethod + def _query_whois(domain: str) -> Domain: + return query(domain) + + @staticmethod + def _prepare_expiration_data( + expiration_date: Optional[datetime.datetime], result: Dict[str, Any] + ) -> Dict[str, Any]: + days_to_expire = None + now = datetime.datetime.now() + if expiration_date: + days_to_expire = (expiration_date - now).days + result["expiration_date"] = expiration_date + if days_to_expire and days_to_expire <= Config.Modules.DomainExpirationScanner.DOMAIN_EXPIRATION_TIMEFRAME_DAYS: + result["close_expiration_date"] = True + result["days_to_expire"] = days_to_expire + return result + + @staticmethod + def _prepare_expiration_status_reason(days_to_expire: int, expiration_date: datetime.datetime) -> str: + return ( + f"Scanned domain will expire in {days_to_expire} days - (on {expiration_date})." + if days_to_expire != 1 + else f"Scanned domain will expire in {days_to_expire} day - (on {expiration_date})." + ) + + +if __name__ == "__main__": + DomainExpirationScanner().loop() diff --git a/artemis/reporting/base/reporter.py b/artemis/reporting/base/reporter.py index 81b0a953b..9fab46419 100644 --- a/artemis/reporting/base/reporter.py +++ b/artemis/reporting/base/reporter.py @@ -2,7 +2,13 @@ from typing import Any, Callable, Dict, List, Tuple from .language import Language -from .normal_form import NormalForm, get_url_normal_form, get_url_score +from .normal_form import ( + NormalForm, + get_domain_normal_form, + get_domain_score, + get_url_normal_form, + get_url_score, +) from .report import Report from .report_type import ReportType from .templating import ReportEmailTemplateFragment @@ -64,13 +70,17 @@ def dict_to_tuple(d: Dict[str, str]) -> Tuple[Tuple[str, str], ...]: @staticmethod def default_scoring_rule(report: Report) -> List[int]: - return [get_url_score(report.target)] + assert report.target_is_url() or report.target_is_domain() + return [get_url_score(report.target) if report.target_is_url() else get_domain_score(report.target)] @staticmethod def default_normal_form_rule(report: Report) -> NormalForm: + assert report.target_is_url() or report.target_is_domain() return Reporter.dict_to_tuple( { "type": report.report_type, - "target": get_url_normal_form(report.target), + "target": get_url_normal_form(report.target) + if report.target_is_url() + else get_domain_normal_form(report.target), } ) diff --git a/artemis/reporting/modules/domain_expiration_scanner/reporter.py b/artemis/reporting/modules/domain_expiration_scanner/reporter.py new file mode 100644 index 000000000..cc81efa7d --- /dev/null +++ b/artemis/reporting/modules/domain_expiration_scanner/reporter.py @@ -0,0 +1,47 @@ +import os +from typing import Any, Dict, List + +from artemis.reporting.base.language import Language +from artemis.reporting.base.report import Report +from artemis.reporting.base.report_type import ReportType +from artemis.reporting.base.reporter import Reporter +from artemis.reporting.base.templating import ReportEmailTemplateFragment +from artemis.reporting.utils import get_top_level_target + + +class DomainExpirationScannerReporter(Reporter): + CLOSE_DOMAIN_EXPIRATION_DATE = ReportType("close_domain_expiration_date") + + @staticmethod + def create_reports(task_result: Dict[str, Any], language: Language) -> List[Report]: + if task_result["headers"]["receiver"] != "domain_expiration_scanner": + return [] + + if not task_result["status"] == "INTERESTING": + return [] + + if not isinstance(task_result["result"], dict): + return [] + + data = task_result["result"] + expiration_date_from_result = data["expiration_date"] + expiration_date = expiration_date_from_result.strftime("%d-%m-%Y") + + return [ + Report( + top_level_target=get_top_level_target(task_result), + target=task_result["payload"]["domain"], + report_type=DomainExpirationScannerReporter.CLOSE_DOMAIN_EXPIRATION_DATE, + additional_data={"expiration_date": expiration_date}, + timestamp=task_result["created_at"], + ) + ] + + @staticmethod + def get_email_template_fragments() -> List[ReportEmailTemplateFragment]: + return [ + ReportEmailTemplateFragment.from_file( + os.path.join(os.path.dirname(__file__), "template_close_domain_expiration_scanner.jinja2"), + priority=5, + ), + ] diff --git a/artemis/reporting/modules/domain_expiration_scanner/template_close_domain_expiration_scanner.jinja2 b/artemis/reporting/modules/domain_expiration_scanner/template_close_domain_expiration_scanner.jinja2 new file mode 100644 index 000000000..ac5ed0a38 --- /dev/null +++ b/artemis/reporting/modules/domain_expiration_scanner/template_close_domain_expiration_scanner.jinja2 @@ -0,0 +1,13 @@ +{% if "close_domain_expiration_date" in data.contains_type %} +
  • {% trans %}The following domains will soon expire: {% endtrans %} + +
    • +{% endif %} diff --git a/artemis/reporting/modules/domain_expiration_scanner/translations/en_US/LC_MESSAGES/messages.po b/artemis/reporting/modules/domain_expiration_scanner/translations/en_US/LC_MESSAGES/messages.po new file mode 100644 index 000000000..dcb86384d --- /dev/null +++ b/artemis/reporting/modules/domain_expiration_scanner/translations/en_US/LC_MESSAGES/messages.po @@ -0,0 +1,7 @@ +#: artemis/reporting/modules/domain_expiration_scanner/template_close_domain_expiration_scanner.jinja2:2 +msgid "The following domains will soon expire: " +msgstr "" + +#: artemis/reporting/modules/domain_expiration_scanner/template_close_domain_expiration_scanner.jinja2:7 +msgid "will expire on" +msgstr "" diff --git a/artemis/reporting/modules/domain_expiration_scanner/translations/pl_PL/LC_MESSAGES/messages.po b/artemis/reporting/modules/domain_expiration_scanner/translations/pl_PL/LC_MESSAGES/messages.po new file mode 100644 index 000000000..9750f118c --- /dev/null +++ b/artemis/reporting/modules/domain_expiration_scanner/translations/pl_PL/LC_MESSAGES/messages.po @@ -0,0 +1,7 @@ +#: artemis/reporting/modules/domain_expiration_scanner/template_close_domain_expiration_scanner.jinja2:2 +msgid "The following domains will soon expire: " +msgstr "Poniższe domeny wkrótce wygasną: " + +#: artemis/reporting/modules/domain_expiration_scanner/template_close_domain_expiration_scanner.jinja2:7 +msgid "will expire on" +msgstr "wygaśnie dnia" diff --git a/artemis/reporting/severity.py b/artemis/reporting/severity.py index 64da9a9cf..786f329a3 100644 --- a/artemis/reporting/severity.py +++ b/artemis/reporting/severity.py @@ -40,6 +40,7 @@ class Severity(str, Enum): ReportType("directory_index"): Severity.MEDIUM, ReportType("open_port_remote_desktop"): Severity.MEDIUM, ReportType("exposed_bash_history"): Severity.MEDIUM, + ReportType("close_domain_expiry_date"): Severity.MEDIUM, ReportType("certificate_authority_invalid"): Severity.LOW, ReportType("expired_ssl_certificate"): Severity.LOW, ReportType("almost_expired_ssl_certificate"): Severity.LOW, diff --git a/docker-compose.yaml b/docker-compose.yaml index 242b9c24a..9a6f41b43 100644 --- a/docker-compose.yaml +++ b/docker-compose.yaml @@ -337,6 +337,16 @@ services: restart: always volumes: ["./docker/karton.ini:/etc/karton/karton.ini", "${DOCKER_COMPOSE_ADDITIONAL_SHARED_DIRECTORY:-./shared}:/shared/"] + karton-domain_expiration_scanner: + build: + context: . + dockerfile: docker/Dockerfile + command: "python3 -m artemis.modules.domain_expiration_scanner" + depends_on: [ karton-logger ] + env_file: .env + restart: always + volumes: [ "./docker/karton.ini:/etc/karton/karton.ini" ] + volumes: data-mongodb: data-redis: diff --git a/docker/Dockerfile b/docker/Dockerfile index 10c2318e9..2d5a0e41d 100644 --- a/docker/Dockerfile +++ b/docker/Dockerfile @@ -3,7 +3,7 @@ FROM python:3.11-alpine3.18 COPY docker/wait-for-it.sh /wait-for-it.sh ARG ADDITIONAL_REQUIREMENTS -RUN apk add --no-cache --virtual .build-deps go gcc git libc-dev make libffi-dev libpcap-dev postgresql-dev && \ +RUN apk add --no-cache --virtual .build-deps go gcc git libc-dev make libffi-dev libpcap-dev postgresql-dev whois && \ apk add --no-cache bash libpcap libpq git subversion RUN GOBIN=/usr/local/bin/ go install github.com/projectdiscovery/naabu/v2/cmd/naabu@v2.1.6 && \ GOBIN=/usr/local/bin/ go install github.com/praetorian-inc/fingerprintx/cmd/fingerprintx@v1.1.9 && \ diff --git a/requirements.txt b/requirements.txt index ac1002d25..ac560ac24 100644 --- a/requirements.txt +++ b/requirements.txt @@ -48,3 +48,4 @@ typing-extensions==4.8.0 urllib3==1.26.17 uvicorn==0.23.2 validators==0.22.0 +whois==0.9.27 diff --git a/test/modules/test_domain_expiration_scanner.py b/test/modules/test_domain_expiration_scanner.py new file mode 100644 index 000000000..b4749ec12 --- /dev/null +++ b/test/modules/test_domain_expiration_scanner.py @@ -0,0 +1,28 @@ +import datetime +from test.base import ArtemisModuleTestCase +from unittest.mock import patch + +from karton.core import Task + +from artemis.binds import TaskStatus, TaskType +from artemis.modules.domain_expiration_scanner import DomainExpirationScanner + + +class TestDomainExpirationScanner(ArtemisModuleTestCase): + karton_class = DomainExpirationScanner # type: ignore + + def test_simple(self) -> None: + task = Task( + {"type": TaskType.DOMAIN.value}, + payload={"domain": "google.com"}, + ) + + with patch("artemis.config.Config.Modules.DomainExpirationScanner") as mocked_config: + mocked_config.DOMAIN_EXPIRATION_TIMEFRAME_DAYS = 5000 + self.run_task(task) + + (call,) = self.mock_db.save_task_result.call_args_list + self.assertEqual(call.kwargs["status"], TaskStatus.INTERESTING) + reason = call.kwargs["status_reason"] + self.assertTrue(reason.startswith("Scanned domain will expire in")) + self.assertTrue(isinstance(call.kwargs["data"]["expiration_date"], datetime.datetime)) diff --git a/test/reporting/test_domain_expiration_scanner_autoreporter_integration.py b/test/reporting/test_domain_expiration_scanner_autoreporter_integration.py new file mode 100644 index 000000000..5ea2af750 --- /dev/null +++ b/test/reporting/test_domain_expiration_scanner_autoreporter_integration.py @@ -0,0 +1,48 @@ +from test.base import BaseReportingTest +from unittest.mock import patch + +from karton.core import Task + +from artemis.binds import TaskType +from artemis.modules.domain_expiration_scanner import DomainExpirationScanner +from artemis.reporting.base.language import Language +from artemis.reporting.base.reporters import reports_from_task_result + + +class DomainExpirationScannerAutoreporterIntegrationTest(BaseReportingTest): + karton_class = DomainExpirationScanner # type: ignore + + def test_domain_expiration(self) -> None: + message = self._run_task_and_get_message("google.com") + self.assertIn("The following domains will soon expire:", message) + self.assertIn("google.com", message) + + def _run_task_and_get_message(self, domain: str) -> str: + task = Task( + {"type": TaskType.DOMAIN}, + payload={TaskType.DOMAIN: domain}, + ) + with patch("artemis.config.Config.Modules.DomainExpirationScanner") as mocked_config: + mocked_config.DOMAIN_EXPIRATION_TIMEFRAME_DAYS = 5000 + self.run_task(task) + (call,) = self.mock_db.save_task_result.call_args_list + data = { + "created_at": None, + "headers": { + "receiver": "domain_expiration_scanner", + }, + "payload": { + "domain": domain, + }, + "payload_persistent": { + "original_domain": domain, + }, + "status": "INTERESTING", + "result": call.kwargs["data"], + } + + reports = reports_from_task_result(data, Language.en_US) + message_template = self.generate_message_template() + return message_template.render( + {"data": {"contains_type": set([report.report_type for report in reports]), "reports": reports}} + )