Merge pull request #183 from rachit367/feat/technology-type-and-tech-detect

feat(types,enrichers): add Technology type and tech_detect transformer

flowsint-enrichers/src/flowsint_enrichers/website/to_technologies.py

@@ -0,0 +1,133 @@
+from typing import Any, Dict, List, Optional
+from flowsint_core.core.logger import Logger
+from flowsint_core.core.enricher_base import Enricher
+from flowsint_enrichers.registry import flowsint_enricher
+from flowsint_types.website import Website
+from flowsint_types.technology import Technology
+from tools.network.httpx import HttpxTool
+
+
+@flowsint_enricher
+class TechDetectEnricher(Enricher):
+    """[HTTPX] Detect web technologies on a website via httpx (wappalyzer)."""
+
+    InputType = Website
+    OutputType = Technology
+
+    def __init__(
+        self,
+        sketch_id: Optional[str] = None,
+        scan_id: Optional[str] = None,
+        vault=None,
+        params: Optional[Dict[str, Any]] = None,
+        **kwargs,
+    ):
+        super().__init__(
+            sketch_id=sketch_id,
+            scan_id=scan_id,
+            params_schema=self.get_params_schema(),
+            vault=vault,
+            params=params,
+            **kwargs,
+        )
+
+    @classmethod
+    def name(cls) -> str:
+        return "tech_detect"
+
+    @classmethod
+    def category(cls) -> str:
+        return "Website"
+
+    @classmethod
+    def key(cls) -> str:
+        return "website"
+
+    @staticmethod
+    def _parse_tech(entry: str) -> Optional[Technology]:
+        """Turn an httpx tech entry into a Technology.
+
+        httpx -td emits plain names ("nginx") and, when wappalyzer knows
+        the version, "name:version" pairs ("PHP:8.1").
+        """
+        entry = (entry or "").strip()
+        if not entry:
+            return None
+        name, sep, version = entry.partition(":")
+        name = name.strip()
+        if not name:
+            return None
+        return Technology(
+            name=name,
+            version=(version.strip() or None) if sep else None,
+            source="httpx",
+        )
+
+    async def scan(self, data: List[InputType]) -> List[OutputType]:
+        results: List[OutputType] = []
+
+        try:
+            httpx = HttpxTool()
+        except Exception as e:
+            Logger.error(
+                self.sketch_id,
+                {"message": f"[HTTPX] Failed to initialize httpx: {e}"},
+            )
+            return results
+
+        for website in data:
+            url = str(website.url)
+            try:
+                probes = httpx.launch(url, args=["-td"])
+            except Exception as e:
+                Logger.error(
+                    self.sketch_id,
+                    {"message": f"[HTTPX] Error probing {url}: {e}"},
+                )
+                continue
+
+            seen: set[tuple[str, Optional[str]]] = set()
+            for probe in probes:
+                for entry in probe.get("tech", []) or []:
+                    tech = self._parse_tech(entry)
+                    if tech is None:
+                        continue
+                    dedup_key = (tech.name, tech.version)
+                    if dedup_key in seen:
+                        continue
+                    seen.add(dedup_key)
+                    # Carry the source website through to postprocess.
+                    setattr(tech, "_source_url", url)
+                    results.append(tech)
+                    Logger.info(
+                        self.sketch_id,
+                        {"message": f"[HTTPX] {url} -> {tech.nodeLabel}"},
+                    )
+
+        return results
+
+    def postprocess(
+        self, results: List[OutputType], original_input: List[InputType] = None
+    ) -> List[OutputType]:
+        for tech in results:
+            if not self._graph_service:
+                continue
+            source_url = getattr(tech, "_source_url", None)
+            if not source_url:
+                continue
+
+            website_obj = Website(url=source_url)
+            self.create_node(website_obj)
+            self.create_node(tech)
+            self.create_relationship(website_obj, tech, "USES_TECHNOLOGY")
+            self.log_graph_message(
+                f"Technology {tech.nodeLabel} detected on {source_url}"
+            )
+
+            delattr(tech, "_source_url")
+
+        return results
+
+
+InputType = TechDetectEnricher.InputType
+OutputType = TechDetectEnricher.OutputType

flowsint-enrichers/tests/enrichers/test_tech_detect.py

@@ -0,0 +1,112 @@
+import pytest
+
+from flowsint_enrichers import ENRICHER_REGISTRY
+from flowsint_enrichers.website.to_technologies import TechDetectEnricher
+from flowsint_types.technology import Technology
+from flowsint_types.website import Website
+
+
+# ---------------------------------------------------------------------------
+# Technology type
+# ---------------------------------------------------------------------------
+def test_technology_label_without_version():
+    assert Technology(name="nginx").nodeLabel == "nginx"
+
+
+def test_technology_label_with_version():
+    assert Technology(name="PHP", version="8.1").nodeLabel == "PHP 8.1"
+
+
+def test_technology_from_string_plain_and_versioned():
+    assert Technology.from_string("nginx").version is None
+    t = Technology.from_string("PHP:8.1")
+    assert (t.name, t.version) == ("PHP", "8.1")
+
+
+def test_technology_is_in_type_registry():
+    from flowsint_types import TYPE_REGISTRY
+
+    assert TYPE_REGISTRY.get_lowercase("technology") is Technology
+
+
+# ---------------------------------------------------------------------------
+# Registry wiring
+# ---------------------------------------------------------------------------
+def test_tech_detect_is_registered():
+    enricher = ENRICHER_REGISTRY.get_enricher("tech_detect", "123", "123")
+    assert enricher.name() == "tech_detect"
+
+
+def test_tech_detect_metadata():
+    assert TechDetectEnricher.category() == "Website"
+    assert TechDetectEnricher.key() == "website"
+    assert TechDetectEnricher.input_schema()["type"] == "Website"
+    assert TechDetectEnricher.output_schema()["type"] == "Technology"
+
+
+# ---------------------------------------------------------------------------
+# _parse_tech
+# ---------------------------------------------------------------------------
+def test_parse_tech_plain():
+    t = TechDetectEnricher._parse_tech("nginx")
+    assert (t.name, t.version, t.source) == ("nginx", None, "httpx")
+
+
+def test_parse_tech_versioned():
+    t = TechDetectEnricher._parse_tech("PHP:8.1")
+    assert (t.name, t.version) == ("PHP", "8.1")
+
+
+def test_parse_tech_blank_returns_none():
+    assert TechDetectEnricher._parse_tech("") is None
+    assert TechDetectEnricher._parse_tech("   ") is None
+    assert TechDetectEnricher._parse_tech(":1.0") is None
+
+
+# ---------------------------------------------------------------------------
+# scan() — httpx mocked, no Docker daemon touched
+# ---------------------------------------------------------------------------
+class _FakeHttpx:
+    def __init__(self, by_url):
+        self._by_url = by_url
+        self.calls = []
+
+    def launch(self, target, args=None):
+        self.calls.append((target, args))
+        return self._by_url.get(target, [])
+
+
+@pytest.mark.asyncio
+async def test_scan_extracts_and_dedupes_technologies(monkeypatch):
+    fake = _FakeHttpx(
+        {
+            "https://example.com/": [
+                {"tech": ["nginx", "PHP:8.1", "nginx"]},  # duplicate nginx
+            ]
+        }
+    )
+    monkeypatch.setattr(
+        "flowsint_enrichers.website.to_technologies.HttpxTool", lambda: fake
+    )
+
+    enricher = TechDetectEnricher(sketch_id="s", scan_id="t", graph_service=None)
+    results = await enricher.scan([Website(url="https://example.com/")])
+
+    labels = sorted(t.nodeLabel for t in results)
+    assert labels == ["PHP 8.1", "nginx"]
+    assert all(isinstance(t, Technology) for t in results)
+    assert all(getattr(t, "_source_url") == "https://example.com/" for t in results)
+    # -td flag forwarded to httpx
+    assert fake.calls == [("https://example.com/", ["-td"])]
+
+
+@pytest.mark.asyncio
+async def test_scan_handles_no_tech_field(monkeypatch):
+    fake = _FakeHttpx({"https://example.com/": [{"status_code": 200}]})
+    monkeypatch.setattr(
+        "flowsint_enrichers.website.to_technologies.HttpxTool", lambda: fake
+    )
+
+    enricher = TechDetectEnricher(sketch_id="s", scan_id="t", graph_service=None)
+    results = await enricher.scan([Website(url="https://example.com/")])
+    assert results == []

flowsint-types/src/flowsint_types/__init__.py

@@ -43,6 +43,7 @@ from .script import Script
 from .session import Session
 from .social_account import SocialAccount
 from .ssl_certificate import SSLCertificate
+from .technology import Technology
 from .username import Username
 from .wallet import CryptoNFT, CryptoWallet, CryptoWalletTransaction
 from .weapon import Weapon
@@ -87,6 +88,7 @@ __all__ = [
     "Session",
     "SocialAccount",
     "SSLCertificate",
+    "Technology",
     "Username",
     "CryptoWallet",
     "CryptoWalletTransaction",
@@ -134,6 +136,7 @@ TYPE_TO_MODEL: Dict[str, Type[BaseModel]] = {
     "file": File,
     "malware": Malware,
     "sslcertificate": SSLCertificate,
+    "technology": Technology,
     "location": Location,
     "affiliation": Affiliation,
     "alias": Alias,

flowsint-types/src/flowsint_types/technology.py

@@ -0,0 +1,57 @@
+from typing import Optional, Self
+
+from pydantic import Field, model_validator
+
+from .flowsint_base import FlowsintType
+from .registry import flowsint_type
+
+
+@flowsint_type
+class Technology(FlowsintType):
+    """Represents a technology, framework, or software detected during recon."""
+
+    name: str = Field(
+        ...,
+        description="Technology name (e.g. nginx, PHP, React)",
+        title="Technology Name",
+        json_schema_extra={"primary": True},
+    )
+    version: Optional[str] = Field(
+        None, description="Detected version, if known", title="Version"
+    )
+    category: Optional[str] = Field(
+        None,
+        description="Technology category (e.g. web-server, framework, cms)",
+        title="Category",
+    )
+    source: Optional[str] = Field(
+        None,
+        description="Tool or method that detected the technology (e.g. httpx)",
+        title="Source",
+    )
+
+    @model_validator(mode="after")
+    def compute_label(self) -> Self:
+        if self.version:
+            self.nodeLabel = f"{self.name} {self.version}"
+        else:
+            self.nodeLabel = self.name
+        return self
+
+    @classmethod
+    def from_string(cls, line: str):
+        """Parse a technology from a raw string.
+
+        Accepts either a bare name ("nginx") or a wappalyzer-style
+        "name:version" pair ("PHP:8.1"), as emitted by httpx -td.
+        """
+        line = line.strip()
+        if ":" in line:
+            name, _, version = line.partition(":")
+            return cls(name=name.strip(), version=version.strip() or None)
+        return cls(name=line)
+
+    @classmethod
+    def detect(cls, line: str) -> bool:
+        """Technology cannot be reliably detected from a single line of text."""
+        return False