Add Xquik SpiderFoot module

Author: kriptoburak

modules/sfp_xquik.py

@@ -0,0 +1,183 @@
+# -------------------------------------------------------------------------------
+# Name:         sfp_xquik
+# Purpose:      Query Xquik for X/Twitter profile information.
+#
+# Author:      SpiderFoot contributors
+#
+# Created:     2026-06-13
+# Copyright:   (c) SpiderFoot contributors 2026
+# Licence:     MIT
+# -------------------------------------------------------------------------------
+
+import json
+import re
+import urllib.parse
+
+from spiderfoot import SpiderFootEvent, SpiderFootPlugin
+
+
+class sfp_xquik(SpiderFootPlugin):
+
+    meta = {
+        'name': "Xquik",
+        'summary': "Gather X/Twitter user profile details from Xquik.",
+        'flags': ["apikey"],
+        'useCases': ["Footprint", "Investigate", "Passive"],
+        'categories': ["Social Media"],
+        'dataSource': {
+            'website': "https://xquik.com/",
+            'model': "FREE_AUTH_LIMITED",
+            'references': [
+                "https://docs.xquik.com/api-reference/overview"
+            ],
+            'apiKeyInstructions': [
+                "Visit https://xquik.com/",
+                "Create or sign in to your account",
+                "Open API keys from the dashboard",
+                "Create an API key and paste it into the module options"
+            ],
+            'favIcon': "https://xquik.com/icon.svg",
+            'logo': "https://xquik.com/icon.svg",
+            'description': "Xquik provides X/Twitter data APIs for profile lookup, "
+            "tweet search, media download, monitoring, and automation workflows.",
+        }
+    }
+
+    opts = {
+        "api_key": "",
+        "base_url": "https://xquik.com/api/v1"
+    }
+
+    optdescs = {
+        "api_key": "Xquik API key.",
+        "base_url": "Xquik API base URL."
+    }
+
+    handle_re = re.compile(r"^[A-Za-z0-9_]{1,15}$")
+
+    results = None
+    errorState = False
+
+    def setup(self, sfc, userOpts=dict()):
+        self.sf = sfc
+        self.__dataSource__ = "Xquik"
+        self.results = self.tempStorage()
+        self.errorState = False
+
+        for opt in list(userOpts.keys()):
+            self.opts[opt] = userOpts[opt]
+
+    def watchedEvents(self):
+        return ["SOCIAL_MEDIA"]
+
+    def producedEvents(self):
+        return ["RAW_RIR_DATA", "USERNAME"]
+
+    def extract_handle(self, event_data):
+        try:
+            network, value = event_data.split(": ", 1)
+        except ValueError:
+            return None
+
+        if network.lower() not in ["twitter", "x"]:
+            return None
+
+        url = value.replace("<SFURL>", "").replace("</SFURL>", "").strip()
+
+        try:
+            parsed = urllib.parse.urlparse(url)
+        except Exception:
+            return None
+
+        if parsed.scheme not in ["http", "https"]:
+            return None
+
+        if parsed.netloc.lower() not in ["twitter.com", "www.twitter.com", "x.com", "www.x.com"]:
+            return None
+
+        parts = [part for part in parsed.path.split("/") if part]
+        if len(parts) != 1:
+            return None
+
+        handle = parts[0].lstrip("@")
+        if not self.handle_re.match(handle):
+            return None
+
+        if handle.lower() in ["home", "explore", "i", "intent", "messages", "notifications", "search"]:
+            return None
+
+        return handle
+
+    def query(self, handle):
+        base_url = self.opts["base_url"].rstrip("/")
+        user = urllib.parse.quote(handle, safe="")
+        headers = {
+            "x-api-key": self.opts["api_key"]
+        }
+
+        return self.sf.fetchUrl(f"{base_url}/x/users/{user}",
+                                timeout=self.opts.get("_fetchtimeout", 30),
+                                useragent="SpiderFoot",
+                                headers=headers)
+
+    def handleEvent(self, event):
+        eventData = event.data
+
+        if self.errorState:
+            return
+
+        handle = self.extract_handle(eventData)
+        if handle is None:
+            return
+
+        lookup_key = handle.lower()
+        if lookup_key in self.results:
+            return
+
+        self.results[lookup_key] = True
+
+        if self.opts["api_key"] == "":
+            self.error("You enabled sfp_xquik but did not set an API key!")
+            self.errorState = True
+            return
+
+        res = self.query(handle)
+        if not res or not res.get("content"):
+            return
+
+        code = str(res.get("code"))
+        if code in ["401", "402", "403"]:
+            self.error(f"Xquik API request failed with HTTP {code}.")
+            self.errorState = True
+            return
+
+        if code != "200":
+            self.debug(f"Xquik API request failed with HTTP {code}.")
+            return
+
+        try:
+            data = json.loads(res["content"])
+        except Exception as e:
+            self.error(f"Error processing JSON response from Xquik: {e}")
+            return
+
+        evt = SpiderFootEvent("RAW_RIR_DATA", json.dumps(data, sort_keys=True),
+                              self.__name__, event)
+        self.notifyListeners(evt)
+
+        payload = data.get("data", data)
+        if not isinstance(payload, dict):
+            return
+
+        username = payload.get("username") or payload.get("screen_name") or payload.get("handle")
+        if not username or not isinstance(username, str):
+            return
+
+        username = username.lstrip("@")
+        if not self.handle_re.match(username):
+            return
+
+        evt = SpiderFootEvent("USERNAME", username, self.__name__, event)
+        self.notifyListeners(evt)
+
+# End of sfp_xquik class

test/unit/modules/test_sfp_xquik.py

@@ -0,0 +1,101 @@
+import json
+import pytest
+import unittest
+
+from modules.sfp_xquik import sfp_xquik
+from sflib import SpiderFoot
+from spiderfoot import SpiderFootEvent, SpiderFootTarget
+
+
+@pytest.mark.usefixtures
+class TestModuleXquik(unittest.TestCase):
+
+    def test_opts(self):
+        module = sfp_xquik()
+        self.assertEqual(len(module.opts), len(module.optdescs))
+
+    def test_setup(self):
+        sf = SpiderFoot(self.default_options)
+        module = sfp_xquik()
+        module.setup(sf, dict())
+
+    def test_watchedEvents_should_return_list(self):
+        module = sfp_xquik()
+        self.assertIsInstance(module.watchedEvents(), list)
+
+    def test_producedEvents_should_return_list(self):
+        module = sfp_xquik()
+        self.assertIsInstance(module.producedEvents(), list)
+
+    def test_extract_handle_should_parse_x_profile_url(self):
+        module = sfp_xquik()
+        result = module.extract_handle("X: <SFURL>https://x.com/xquik</SFURL>")
+        self.assertEqual(result, "xquik")
+
+    def test_extract_handle_should_skip_status_url(self):
+        module = sfp_xquik()
+        result = module.extract_handle("Twitter: <SFURL>https://twitter.com/xquik/status/1</SFURL>")
+        self.assertIsNone(result)
+
+    def test_handleEvent_no_api_key_should_set_errorState(self):
+        sf = SpiderFoot(self.default_options)
+
+        module = sfp_xquik()
+        module.setup(sf, dict())
+
+        target_value = 'spiderfoot.net'
+        target_type = 'INTERNET_NAME'
+        target = SpiderFootTarget(target_value, target_type)
+        module.setTarget(target)
+
+        event_type = 'SOCIAL_MEDIA'
+        event_data = 'Twitter: <SFURL>https://twitter.com/xquik</SFURL>'
+        event_module = 'example module'
+        source_event = SpiderFootEvent('ROOT', target_value, '', '')
+        evt = SpiderFootEvent(event_type, event_data, event_module, source_event)
+
+        result = module.handleEvent(evt)
+
+        self.assertIsNone(result)
+        self.assertTrue(module.errorState)
+
+    def test_handleEvent_should_emit_raw_data_and_username(self):
+        sf = SpiderFoot(self.default_options)
+
+        module = sfp_xquik()
+        module.setup(sf, {"api_key": "test-key"})
+
+        target_value = 'spiderfoot.net'
+        target_type = 'INTERNET_NAME'
+        target = SpiderFootTarget(target_value, target_type)
+        module.setTarget(target)
+
+        captured = list()
+
+        def new_fetchUrl(url, timeout=0, useragent=None, headers=None):
+            self.assertEqual(url, "https://xquik.com/api/v1/x/users/xquik")
+            self.assertEqual(timeout, 30)
+            self.assertEqual(useragent, "SpiderFoot")
+            self.assertEqual(headers, {"x-api-key": "test-key"})
+            return {
+                "code": "200",
+                "content": json.dumps({"data": {"username": "xquik", "followersCount": 42}})
+            }
+
+        def new_notifyListeners(self, event):
+            captured.append(event)
+
+        sf.fetchUrl = new_fetchUrl
+        module.notifyListeners = new_notifyListeners.__get__(module, sfp_xquik)
+
+        event_type = 'SOCIAL_MEDIA'
+        event_data = 'Twitter: <SFURL>https://twitter.com/xquik</SFURL>'
+        event_module = 'example module'
+        source_event = SpiderFootEvent('ROOT', target_value, '', '')
+        evt = SpiderFootEvent(event_type, event_data, event_module, source_event)
+
+        result = module.handleEvent(evt)
+
+        self.assertIsNone(result)
+        self.assertEqual([event.eventType for event in captured], ["RAW_RIR_DATA", "USERNAME"])
+        self.assertEqual(captured[1].data, "xquik")