Report that the domain will soon expire #495

artemis/config.py

@@ -404,6 +404,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_ALERT_IN_DAYS", default=5, cast=int)
+
     @staticmethod
     def verify_each_variable_is_annotated() -> None:
         def verify_class(cls: type) -> None:

artemis/modules/domain_expiration_scanner.py

@@ -0,0 +1,56 @@
+#!/usr/bin/env python3
+import datetime
+from typing import Any, Dict
+
+from karton.core import Task
+from whois import 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] = {}
+        if is_main_domain(domain):
+            now = datetime.datetime.now()
+            domain_data = query(domain)
+            expiry_date = domain_data.expiration_date
+            days_to_expire = None
+            if expiry_date:
+                days_to_expire = (expiry_date - now).days
+            result["expiration_date"] = expiry_date
+            if (
+                days_to_expire
+                and days_to_expire <= Config.Modules.DomainExpirationScanner.DOMAIN_EXPIRATION_TIMEFRAME_DAYS
+            ):
+                result["close_expiry_date"] = True
+                result["days_to_expire"] = days_to_expire
+                status = TaskStatus.INTERESTING
+                status_reason = (
+                    f"Scanned domain will expire in {days_to_expire} days - {expiry_date}."
+                    if days_to_expire != 1
+                    else f"Scanned domain will expire in {days_to_expire} day - (on {expiry_date})."
+                )
+            else:
+                status = TaskStatus.OK
+                status_reason = None
+                self.db.save_task_result(task=current_task, status=status, status_reason=status_reason, data=result)
+        else:
+            status = TaskStatus.OK
+            status_reason = None
+        self.db.save_task_result(task=current_task, status=status, status_reason=status_reason, data=result)
+
+
+if __name__ == "__main__":
+    DomainExpirationScanner().loop()

artemis/reporting/modules/domain_expiration_scanner/reporter.py

@@ -0,0 +1,74 @@
+import os
+from typing import Any, Callable, Dict, List
+
+from artemis.reporting.base.language import Language
+from artemis.reporting.base.normal_form import (
+    NormalForm,
+    get_domain_normal_form,
+    get_domain_score,
+)
+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,
+            ),
+        ]
+
+    @staticmethod
+    def get_scoring_rules() -> Dict[ReportType, Callable[[Report], List[int]]]:
+        """See the docstring in the parent class."""
+        return {
+            DomainExpirationScannerReporter.CLOSE_DOMAIN_EXPIRATION_DATE: lambda report: [
+                get_domain_score(report.target)
+            ]
+        }
+
+    @staticmethod
+    def get_normal_form_rules() -> Dict[ReportType, Callable[[Report], NormalForm]]:
+        """See the docstring in the Reporter class."""
+        return {
+            DomainExpirationScannerReporter.CLOSE_DOMAIN_EXPIRATION_DATE: lambda report: Reporter.dict_to_tuple(
+                {
+                    "type": report.report_type,
+                    "target": get_domain_normal_form(report.target),
+                    "message": report.additional_data["expiration_date"],
+                }
+            )
+        }

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 %}
+    <li>{% trans %}The following domains will soon expire: {% endtrans %}
+        <ul>
+            {% for report in data.reports %}
+                {% if report.report_type == "close_domain_expiration_date" %}
+                    <li>
+                        <p> {{ report.target }} - {% trans %}will expire on{% endtrans %} {{ report.additional_data["expiration_date"] }} </p>
+                    </li>
+                {% endif %}
+            {% endfor %}
+        </ul>
+    </li>
+{% endif %}

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 ""

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"

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,

docker-compose.yaml

@@ -335,6 +335,16 @@ services:
     restart: always
     volumes: ["./docker/karton.ini:/etc/karton/karton.ini"]
 
+  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:

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/[email protected] && \
     GOBIN=/usr/local/bin/ go install github.com/praetorian-inc/fingerprintx/cmd/[email protected] && \

requirements.txt

@@ -48,3 +48,4 @@ typing-extensions==4.8.0
 urllib3==1.26.16
 uvicorn==0.23.2
 validators==0.22.0
+whois==0.9.27

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))

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}}
+            )