From 63a97a76f77c60cf6210358a40ee892fe7b75e99 Mon Sep 17 00:00:00 2001 From: rachit367 Date: Tue, 9 Jun 2026 10:45:43 +0530 Subject: [PATCH] feat(types,enrichers): add Technology type and tech_detect transformer Adds application-level technology fingerprinting, extending coverage beyond ip/port/website/ssl_certificate as requested in #59. - New `Technology` type (name, version, category, source) registered in the type registry, the package exports, and TYPE_TO_MODEL. Its from_string accepts both bare names ("nginx") and wappalyzer-style "name:version" pairs ("PHP:8.1"). - New `tech_detect` enricher: Website -> Technology, using the existing httpx tool with -td (wappalyzer dataset). Each detected technology becomes a Technology node linked to the website with a USES_TECHNOLOGY relationship; results are de-duplicated per probe. Tests cover the Technology type (label, from_string, registry), the tech entry parser, enricher registration/metadata, and scan() with the httpx Docker tool mocked, so no Docker daemon is required. Closes #59 --- .../website/to_technologies.py | 133 ++++++++++++++++++ .../tests/enrichers/test_tech_detect.py | 112 +++++++++++++++ flowsint-types/src/flowsint_types/__init__.py | 3 + .../src/flowsint_types/technology.py | 57 ++++++++ 4 files changed, 305 insertions(+) create mode 100644 flowsint-enrichers/src/flowsint_enrichers/website/to_technologies.py create mode 100644 flowsint-enrichers/tests/enrichers/test_tech_detect.py create mode 100644 flowsint-types/src/flowsint_types/technology.py diff --git a/flowsint-enrichers/src/flowsint_enrichers/website/to_technologies.py b/flowsint-enrichers/src/flowsint_enrichers/website/to_technologies.py new file mode 100644 index 00000000..a4e4b0cc --- /dev/null +++ b/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 diff --git a/flowsint-enrichers/tests/enrichers/test_tech_detect.py b/flowsint-enrichers/tests/enrichers/test_tech_detect.py new file mode 100644 index 00000000..6541d671 --- /dev/null +++ b/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 == [] diff --git a/flowsint-types/src/flowsint_types/__init__.py b/flowsint-types/src/flowsint_types/__init__.py index 7f913037..60f6c5db 100644 --- a/flowsint-types/src/flowsint_types/__init__.py +++ b/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, diff --git a/flowsint-types/src/flowsint_types/technology.py b/flowsint-types/src/flowsint_types/technology.py new file mode 100644 index 00000000..1a898563 --- /dev/null +++ b/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