Phunter Analyzer

api_app/analyzers_manager/classes.py

@@ -399,7 +399,11 @@ def _raise_container_not_running(self) -> None:
         )
 
     def _docker_run(
-        self, req_data: dict, req_files: dict = None, analyzer_name: str = None
+        self,
+        req_data: dict,
+        req_files: dict = None,
+        analyzer_name: str = None,
+        avoid_polling: bool = False,
     ) -> dict:
         """
         Helper function that takes of care of requesting new analysis,
@@ -433,8 +437,8 @@ def _docker_run(
             self._raise_container_not_running()
 
         # step #2: raise AnalyzerRunException in case of error
-        # Modified to support synchronous analyzer BBOT that return results directly in the initial response, avoiding unnecessary polling.
-        if analyzer_name == "BBOT_Analyzer":
+        # Modified to support synchronous analyzers that return results directly in the initial response, avoiding unnecessary polling.
+        if avoid_polling:
             report = resp1.json().get("report", None)
             err = resp1.json().get("error", None)
         else:

api_app/analyzers_manager/migrations/0157_analyzer_config_phunter.py

@@ -0,0 +1,125 @@
+from django.db import migrations
+from django.db.models.fields.related_descriptors import (
+    ForwardManyToOneDescriptor,
+    ForwardOneToOneDescriptor,
+    ManyToManyDescriptor,
+    ReverseManyToOneDescriptor,
+    ReverseOneToOneDescriptor,
+)
+
+plugin = {
+    "python_module": {
+        "health_check_schedule": None,
+        "update_schedule": None,
+        "module": "phunter.PhunterAnalyzer",
+        "base_path": "api_app.analyzers_manager.observable_analyzers",
+    },
+    "name": "Phunter",
+    "description": "[Phunter Analyzer](https://github.com/N0rz3/Phunter) is an OSINT tool for finding information about a phone number.",
+    "disabled": False,
+    "soft_time_limit": 60,
+    "routing_key": "default",
+    "health_check_status": True,
+    "type": "observable",
+    "docker_based": True,
+    "maximum_tlp": "RED",
+    "observable_supported": ["generic"],
+    "supported_filetypes": [],
+    "run_hash": False,
+    "run_hash_type": "",
+    "not_supported_filetypes": [],
+    "mapping_data_model": {},
+    "model": "analyzers_manager.AnalyzerConfig",
+}
+
+params = []
+
+values = []
+
+
+def _get_real_obj(Model, field, value):
+    def _get_obj(Model, other_model, value):
+        if isinstance(value, dict):
+            real_vals = {}
+            for key, real_val in value.items():
+                real_vals[key] = _get_real_obj(other_model, key, real_val)
+            value = other_model.objects.get_or_create(**real_vals)[0]
+        # it is just the primary key serialized
+        else:
+            if isinstance(value, int):
+                if Model.__name__ == "PluginConfig":
+                    value = other_model.objects.get(name=plugin["name"])
+                else:
+                    value = other_model.objects.get(pk=value)
+            else:
+                value = other_model.objects.get(name=value)
+        return value
+
+    if (
+        type(getattr(Model, field))
+        in [
+            ForwardManyToOneDescriptor,
+            ReverseManyToOneDescriptor,
+            ReverseOneToOneDescriptor,
+            ForwardOneToOneDescriptor,
+        ]
+        and value
+    ):
+        other_model = getattr(Model, field).get_queryset().model
+        value = _get_obj(Model, other_model, value)
+    elif type(getattr(Model, field)) in [ManyToManyDescriptor] and value:
+        other_model = getattr(Model, field).rel.model
+        value = [_get_obj(Model, other_model, val) for val in value]
+    return value
+
+
+def _create_object(Model, data):
+    mtm, no_mtm = {}, {}
+    for field, value in data.items():
+        value = _get_real_obj(Model, field, value)
+        if type(getattr(Model, field)) is ManyToManyDescriptor:
+            mtm[field] = value
+        else:
+            no_mtm[field] = value
+    try:
+        o = Model.objects.get(**no_mtm)
+    except Model.DoesNotExist:
+        o = Model(**no_mtm)
+        o.full_clean()
+        o.save()
+        for field, value in mtm.items():
+            attribute = getattr(o, field)
+            if value is not None:
+                attribute.set(value)
+        return False
+    return True
+
+
+def migrate(apps, schema_editor):
+    Parameter = apps.get_model("api_app", "Parameter")
+    PluginConfig = apps.get_model("api_app", "PluginConfig")
+    python_path = plugin.pop("model")
+    Model = apps.get_model(*python_path.split("."))
+    if not Model.objects.filter(name=plugin["name"]).exists():
+        exists = _create_object(Model, plugin)
+        if not exists:
+            for param in params:
+                _create_object(Parameter, param)
+            for value in values:
+                _create_object(PluginConfig, value)
+
+
+def reverse_migrate(apps, schema_editor):
+    python_path = plugin.pop("model")
+    Model = apps.get_model(*python_path.split("."))
+    Model.objects.get(name=plugin["name"]).delete()
+
+
+class Migration(migrations.Migration):
+    atomic = False
+    dependencies = [
+        ("api_app", "0071_delete_last_elastic_report"),
+        ("analyzers_manager", "0156_alter_analyzer_config_required_api_key_abuse_ch"),
+    ]
+
+    operations = [migrations.RunPython(migrate, reverse_migrate)]

api_app/analyzers_manager/observable_analyzers/bbot.py

@@ -54,7 +54,9 @@ def run(self):
         logger.info(f"Sending {self.name} scan request: {req_data} to {self.url}")
 
         try:
-            report = self._docker_run(req_data, analyzer_name=self.name)
+            report = self._docker_run(
+                req_data, analyzer_name=self.name, avoid_polling=True
+            )
             logger.info(f"BBOT scan completed successfully with report: {report}")
             return report
         except requests.RequestException as e:

api_app/analyzers_manager/observable_analyzers/phunter.py

@@ -0,0 +1,75 @@
+import logging
+
+import phonenumbers
+import requests
+
+from api_app.analyzers_manager.classes import DockerBasedAnalyzer, ObservableAnalyzer
+from api_app.analyzers_manager.exceptions import AnalyzerRunException
+from tests.mock_utils import MockUpResponse
+
+logging.basicConfig(level=logging.DEBUG)
+logger = logging.getLogger(__name__)
+
+
+class PhunterAnalyzer(ObservableAnalyzer, DockerBasedAnalyzer):
+    name: str = "Phunter"
+    url: str = "http://phunter:5612/analyze"
+    max_tries: int = 1
+    poll_distance: int = 0
+
+    def run(self):
+        try:
+            parsed_number = phonenumbers.parse(self.observable_name)
+
+            formatted_number = phonenumbers.format_number(
+                parsed_number, phonenumbers.PhoneNumberFormat.E164
+            )
+        except phonenumbers.phonenumberutil.NumberParseException:
+            logger.error(f"Phone number parsing failed for: {self.observable_name}")
+            return {"success": False, "error": "Invalid phone number"}
+
+        req_data = {"phone_number": formatted_number}
+        logger.info(f"Sending {self.name} scan request: {req_data} to {self.url}")
+
+        try:
+            response = self._docker_run(
+                req_data, analyzer_name=self.name, avoid_polling=True
+            )
+            logger.info(f"[{self.name}] Scan successful by Phunter. Result: {response}")
+            return response
+
+        except requests.exceptions.RequestException as e:
+            raise AnalyzerRunException(
+                f"[{self.name}] Request failed due to network issue: {e}"
+            )
+
+        except ValueError as e:
+            raise AnalyzerRunException(f"[{self.name}] Invalid response format: {e}")
+
+        except Exception as e:
+            raise AnalyzerRunException(f"{self.name} An unexpected error occurred: {e}")
+
+    @classmethod
+    def update(self):
+        pass
+
+    @staticmethod
+    def mocked_docker_analyzer_post(*args, **kwargs):
+        mock_response = {
+            "success": True,
+            "report": {
+                "valid": "yes",
+                "views": "9",
+                "carrier": "Vodafone",
+                "location": "India",
+                "operator": "Vodafone",
+                "possible": "yes",
+                "line_type": "FIXED LINE OR MOBILE",
+                "local_time": "21:34:45",
+                "spam_status": "Not spammer",
+                "phone_number": "+911234567890",
+                "national_format": "01234567890",
+                "international_format": "+91 1234567890",
+            },
+        }
+        return MockUpResponse(mock_response, 200)

integrations/phunter/Dockerfile

@@ -0,0 +1,24 @@
+FROM python:3.12-slim
+
+# Install dependencies
+RUN apt-get update && apt-get install -y --no-install-recommends git
+
+# Clone Phunter
+RUN git clone https://github.com/N0rz3/Phunter.git /app/Phunter
+
+# Set working directory
+WORKDIR /app
+
+# Copy requirements file and app.py to the working directory
+COPY requirements.txt app.py ./
+
+# Upgrade pip and install Python packages
+RUN pip install --no-cache-dir --upgrade pip && \
+    pip install --no-cache-dir -r requirements.txt && \
+    pip install --no-cache-dir -r /app/Phunter/requirements.txt
+
+# Expose port
+EXPOSE 5612
+
+# Run the app
+CMD ["python", "app.py"]

integrations/phunter/app.py

@@ -0,0 +1,104 @@
+import logging
+import re
+import shlex
+import subprocess
+
+from flask import Flask, jsonify, request
+
+# Logging Configuration
+logging.basicConfig(level=logging.DEBUG)
+logger = logging.getLogger(__name__)
+
+app = Flask(__name__)
+
+
+def strip_ansi_codes(text):
+    """Remove ANSI escape codes from terminal output"""
+    return re.sub(r"\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])", "", text)
+
+
+def parse_phunter_output(output):
+    """Parse output from Phunter CLI and convert to structured JSON"""
+    result = {}
+    key_mapping = {
+        "phone number:": "phone_number",
+        "possible:": "possible",
+        "valid:": "valid",
+        "operator:": "operator",
+        "possible location:": "location",
+        "location:": "location",
+        "carrier:": "carrier",
+        "line type:": "line_type",
+        "international:": "international_format",
+        "national:": "national_format",
+        "local time:": "local_time",
+        "views count:": "views",
+    }
+
+    lines = output.splitlines()
+
+    for line in lines:
+        line = line.strip().lower()
+
+        if "not spammer" in line:
+            result["spam_status"] = "Not spammer"
+            continue
+
+        for keyword, key in key_mapping.items():
+            if keyword in line:
+                value = line.partition(":")[2].strip()
+                if key in ("possible", "valid"):
+                    result[key] = "yes" if "✔" in value else "no"
+                else:
+                    result[key] = value
+                break
+
+    return result
+
+
+@app.route("/analyze", methods=["POST"])
+def analyze():
+    data = request.get_json()
+    phone_number = data.get("phone_number")
+
+    logger.info("Received analysis request")
+
+    if not phone_number:
+        logger.warning("No phone number provided in request")
+        return jsonify({"error": "No phone number provided"}), 400
+
+    try:
+        logger.info("Executing Phunter CLI tool")
+        command_str = f"python3 phunter.py -t {phone_number}"
+        command = shlex.split(command_str)
+        result = subprocess.run(
+            command,
+            capture_output=True,
+            text=True,
+            check=True,
+            cwd="/app/Phunter",
+        )
+
+        raw_output = result.stdout
+        clean_output = strip_ansi_codes(raw_output)
+        parsed_output = parse_phunter_output(clean_output)
+
+        logger.info("Phunter analysis completed")
+
+        return (
+            jsonify(
+                {
+                    "success": True,
+                    "report": parsed_output,
+                }
+            ),
+            200,
+        )
+
+    except subprocess.CalledProcessError as e:
+        return jsonify({"error": f"Phunter execution failed with error {e}"}), 500
+
+
+if __name__ == "__main__":
+    logger.info("Starting Phunter Flask API...")
+    app.run(host="0.0.0.0", port=5612)

integrations/phunter/compose-tests.yml

@@ -0,0 +1,6 @@
+services:
+  phunter:
+    build:
+      context: ../integrations/phunter
+      dockerfile: Dockerfile
+    image: intelowlproject/intelowl_phunter:test

integrations/phunter/compose.yml

@@ -0,0 +1,11 @@
+services:
+  phunter:
+    image: intelowlproject/intelowl_phunter:${REACT_APP_INTELOWL_VERSION}
+    container_name: intelowl_phunter
+    restart: unless-stopped
+    expose:
+      - "5612"
+    volumes:
+      - generic_logs:/var/log/intel_owl
+    depends_on:
+      - uwsgi

integrations/phunter/requirements.txt

@@ -0,0 +1 @@
+flask==3.1.0

requirements/project-requirements.txt

@@ -89,6 +89,7 @@ pylnk3==0.4.2
 androguard==3.4.0a1 # version >=4.x of androguard raises a dependency conflict with quark-engine==25.1.1
 wad==0.4.6
 debloat==1.6.4
+phonenumbers==9.0.3
 
 # httpx required for HTTP/2 support (Mullvad DNS rejects HTTP/1.1 with protocol errors)
 httpx[http2]==0.28.1