Merge pull request #530 from kriptoburak/codex/add-xquik-osint-source

feat(news): add opt-in Xquik search source

.env.example

@@ -33,6 +33,15 @@ AIS_API_KEY=
 # TELEGRAM_OSINT_TRANSLATE=true
 # TELEGRAM_OSINT_TRANSLATE_TO=en
 
+# Xquik search for public X posts. Default off. NEWS_ENABLED=false also stops it.
+# Each poll may consume API credits. Configure one focused search query.
+# XQUIK_ENABLED=false
+# XQUIK_API_KEY=
+# XQUIK_SEARCH_QUERY=
+# XQUIK_SEARCH_LIMIT=20
+# XQUIK_SEARCH_INTERVAL_MINUTES=30
+# XQUIK_SEARCH_TIMEOUT_S=10
+
 # Strategic Risk Analytics (experimental derived OSINT — off by default)
 # GT_ANALYTICS_ENABLED=false
 # GT_ANALYTICS_PROFILE=lean

.github/workflows/ci.yml

@@ -83,4 +83,5 @@ jobs:
             tests/test_liveuamap_parser.py \
             tests/test_liveuamap_provider.py \
             tests/test_liveuamap_docker_contract.py \
+            tests/test_xquik_news.py \
             -v --tb=short

README.md

@@ -636,6 +636,7 @@ ShadowBroker v0.9.7 is composed of three vertically-stacked planes — the **Ope
 | [DigiTraffic](https://www.digitraffic.fi) | European rail positions | ~60s | No |
 | [Global Fishing Watch](https://globalfishingwatch.org) | Fishing vessel activity events | ~1hr | **Yes** (`GFW_API_TOKEN`) |
 | [Telegram public previews](https://t.me/s) | War/OSINT channel posts (`telegram_osint`) | ~1hr | No (optional `TELEGRAM_OSINT_CHANNELS`) |
+| [Xquik](https://docs.xquik.com) | Public X posts from an operator-defined search | ~30min | **Yes** (opt-in) |
 | Transport for London, NYC DOT, TxDOT | CCTV cameras (UK, US) | ~10min | No |
 | Caltrans, WSDOT, GDOT, IDOT, MDOT | CCTV cameras (5 US states) | ~10min | No |
 | Spain DGT, Madrid City | CCTV cameras (Spain) | ~10min | No |
@@ -1141,6 +1142,9 @@ OPENCLAW_ACCESS_TIER=restricted               # OpenClaw agent tier: "restricted
 GFW_API_TOKEN=your_gfw_token                  # Global Fishing Watch — fishing_activity layer (Settings → Maritime)
 TELEGRAM_OSINT_ENABLED=true                   # Telegram OSINT layer (default on)
 TELEGRAM_OSINT_CHANNELS=osintdefender,...     # Comma-separated public channel slugs (see .env.example)
+XQUIK_ENABLED=false                           # Opt-in X search enrichment
+XQUIK_API_KEY=                                # Server-side API key
+XQUIK_SEARCH_QUERY=                           # Operator-defined search query
 
 # Private-lane privacy-core pinning (required when Arti or RNS is enabled)
 PRIVACY_CORE_MIN_VERSION=0.1.0

backend/.env.example

@@ -100,6 +100,15 @@ AIS_API_KEY=              # https://aisstream.io/ — free tier WebSocket key
 # configured news feeds (kill switch for the news layer).
 # NEWS_ENABLED=true
 
+# Xquik search for public X posts. Default off. NEWS_ENABLED=false also stops it.
+# Each poll may consume API credits. Configure one focused search query.
+# XQUIK_ENABLED=false
+# XQUIK_API_KEY=
+# XQUIK_SEARCH_QUERY=
+# XQUIK_SEARCH_LIMIT=20
+# XQUIK_SEARCH_INTERVAL_MINUTES=30
+# XQUIK_SEARCH_TIMEOUT_S=10
+
 # Global Fishing Watch — fishing vessel activity events (Fishing Activity map layer).
 # Free API token from https://globalfishingwatch.org/our-apis/tokens
 # Without this the fishing_activity layer stays empty.

backend/services/api_settings.py

@@ -155,6 +155,15 @@ API_REGISTRY = [
         "url": None,
         "required": False,
     },
+    {
+        "id": "xquik_api_key",
+        "env_key": "XQUIK_API_KEY",
+        "name": "Xquik — API Key",
+        "description": "Server-side key for opt-in X post search in the shared news and threat feed.",
+        "category": "Intelligence",
+        "url": "https://docs.xquik.com/",
+        "required": False,
+    },
     {
         "id": "yfinance",
         "env_key": None,

backend/services/fetchers/news.py

@@ -5,11 +5,13 @@ import time
 import logging
 import calendar
 import concurrent.futures
+
 import requests
 import feedparser
 from services.network_utils import fetch_with_curl
 from services.fetchers._store import latest_data, _data_lock, _mark_fresh
 from services.fetchers.retry import with_retry
+from services.fetchers.xquik_news import fetch_xquik_entries
 from services.oracle_service import enrich_news_items, compute_global_threat_level, detect_breaking_events
 
 
@@ -200,18 +202,23 @@ def fetch_news():
         source_name, url = item
         try:
             xml_data = fetch_with_curl(url, timeout=10).text
-            return source_name, feedparser.parse(xml_data)
+            return source_name, feedparser.parse(xml_data).entries
         except (requests.RequestException, ConnectionError, TimeoutError, ValueError, KeyError, OSError) as e:
             logger.warning(f"Feed {source_name} failed: {e}")
-            return source_name, None
+            return source_name, []
 
     with concurrent.futures.ThreadPoolExecutor(max_workers=min(len(feeds), 6)) as pool:
         feed_results = list(pool.map(_fetch_feed, feeds.items()))
 
-    for source_name, feed in feed_results:
-        if not feed:
+    for entry in fetch_xquik_entries():
+        source_name = entry["source"]
+        source_weights[source_name] = 3
+        feed_results.append((source_name, [entry]))
+
+    for source_name, entries in feed_results:
+        if not entries:
             continue
-        for entry in feed.entries[:5]:
+        for entry in entries[:5]:
             # Drop articles older than the max-age threshold so the
             # threat feed doesn't show stale stories across cycles.
             pp = entry.get("published_parsed")

backend/services/fetchers/xquik_news.py

@@ -0,0 +1,163 @@
+"""Opt-in X post search for the shared news and threat feed."""
+
+from __future__ import annotations
+
+import logging
+import os
+import re
+import threading
+import time
+from datetime import datetime, timezone
+from typing import Any
+
+import requests
+
+from services.network_utils import outbound_user_agent
+
+
+logger = logging.getLogger("services.data_fetcher")
+
+_SEARCH_URL = "https://xquik.com/api/v1/x/tweets/search"
+_TWEET_ID_RE = re.compile(r"^[0-9]{1,32}$")
+_USERNAME_RE = re.compile(r"^[A-Za-z0-9_]{1,15}$")
+_MAX_RESULTS = 100
+_MAX_TEXT_LENGTH = 500
+
+_cache_lock = threading.Lock()
+_cache_signature: tuple[str, int] | None = None
+_cache_entries: list[dict[str, Any]] = []
+_cache_fetched_at = 0.0
+
+
+def xquik_fetch_enabled() -> bool:
+    """Return whether the operator enabled Xquik search enrichment."""
+    return str(os.environ.get("XQUIK_ENABLED", "false")).strip().lower() in {
+        "1",
+        "true",
+        "yes",
+        "on",
+    }
+
+
+def _bounded_int(name: str, default: int, minimum: int, maximum: int) -> int:
+    try:
+        value = int(str(os.environ.get(name, default)).strip())
+    except (TypeError, ValueError):
+        return default
+    return max(minimum, min(maximum, value))
+
+
+def _published_parts(value: object) -> time.struct_time | None:
+    if not isinstance(value, str) or not value.strip():
+        return None
+    try:
+        parsed = datetime.fromisoformat(value.strip().replace("Z", "+00:00"))
+    except ValueError:
+        return None
+    if parsed.tzinfo is None:
+        return None
+    return parsed.astimezone(timezone.utc).timetuple()
+
+
+def _normalize_tweet(tweet: object) -> dict[str, Any] | None:
+    if not isinstance(tweet, dict):
+        return None
+    tweet_id = str(tweet.get("id") or "").strip()
+    text = " ".join(str(tweet.get("text") or "").split())
+    author = tweet.get("author")
+    username = str(author.get("username") or "").strip() if isinstance(author, dict) else ""
+    if not _TWEET_ID_RE.fullmatch(tweet_id) or not _USERNAME_RE.fullmatch(username) or not text:
+        return None
+
+    created_at = tweet.get("createdAt")
+    published_parts = _published_parts(created_at)
+    if published_parts is None:
+        return None
+    entry: dict[str, Any] = {
+        "title": text[:_MAX_TEXT_LENGTH],
+        "summary": "",
+        "link": f"https://x.com/{username}/status/{tweet_id}",
+        "published": created_at,
+        "published_parsed": published_parts,
+        "source": f"Xquik/@{username}",
+    }
+    return entry
+
+
+def _request_entries(api_key: str, query: str, limit: int) -> list[dict[str, Any]] | None:
+    timeout = _bounded_int("XQUIK_SEARCH_TIMEOUT_S", 10, 1, 30)
+    try:
+        response = requests.get(
+            _SEARCH_URL,
+            headers={
+                "User-Agent": outbound_user_agent("xquik-search"),
+                "x-api-key": api_key,
+            },
+            params={
+                "q": query,
+                "queryType": "Latest",
+                "limit": limit,
+                "replies": "exclude",
+                "retweets": "exclude",
+            },
+            timeout=(5, timeout),
+            allow_redirects=False,
+        )
+        if 300 <= response.status_code < 400:
+            raise requests.HTTPError("Xquik search redirected")
+        response.raise_for_status()
+        payload = response.json()
+    except (requests.RequestException, ValueError) as exc:
+        logger.warning("Xquik search failed: %s", type(exc).__name__)
+        return None
+
+    tweets = payload.get("tweets") if isinstance(payload, dict) else None
+    if not isinstance(tweets, list):
+        logger.warning("Xquik search returned an invalid response")
+        return None
+
+    entries = [entry for tweet in tweets[:limit] if (entry := _normalize_tweet(tweet))]
+    logger.info("Xquik search returned %d usable posts", len(entries))
+    return entries
+
+
+def fetch_xquik_entries() -> list[dict[str, Any]]:
+    """Return cached, normalized X posts for the configured search query."""
+    if not xquik_fetch_enabled():
+        return []
+
+    api_key = str(os.environ.get("XQUIK_API_KEY", "")).strip()
+    query = str(os.environ.get("XQUIK_SEARCH_QUERY", "")).strip()
+    if not api_key or not query:
+        logger.warning("Xquik search requires XQUIK_API_KEY and XQUIK_SEARCH_QUERY")
+        return []
+
+    limit = _bounded_int("XQUIK_SEARCH_LIMIT", 20, 1, _MAX_RESULTS)
+    interval_seconds = _bounded_int("XQUIK_SEARCH_INTERVAL_MINUTES", 30, 5, 1440) * 60
+    signature = (query, limit)
+
+    global _cache_entries, _cache_fetched_at, _cache_signature
+    with _cache_lock:
+        if (
+            _cache_signature == signature
+            and _cache_fetched_at
+            and time.monotonic() - _cache_fetched_at < interval_seconds
+        ):
+            return [dict(entry) for entry in _cache_entries]
+
+        entries = _request_entries(api_key, query, limit)
+        if entries is not None:
+            _cache_signature = signature
+            _cache_entries = entries
+            _cache_fetched_at = time.monotonic()
+        if _cache_signature == signature:
+            return [dict(entry) for entry in _cache_entries]
+        return []
+
+
+def _reset_cache_for_tests() -> None:
+    global _cache_entries, _cache_fetched_at, _cache_signature
+    with _cache_lock:
+        _cache_signature = None
+        _cache_entries = []
+        _cache_fetched_at = 0.0

backend/tests/test_xquik_news.py

@@ -0,0 +1,259 @@
+"""Xquik search enrichment for the shared news and threat feed."""
+
+from __future__ import annotations
+
+import time
+from pathlib import Path
+from types import SimpleNamespace
+
+import requests
+
+from services.fetchers import xquik_news
+
+
+class _Response:
+    def __init__(self, payload: object, status_code: int = 200) -> None:
+        self._payload = payload
+        self.status_code = status_code
+
+    def raise_for_status(self) -> None:
+        if self.status_code >= 400:
+            raise requests.HTTPError(f"HTTP {self.status_code}")
+
+    def json(self) -> object:
+        if isinstance(self._payload, ValueError):
+            raise self._payload
+        return self._payload
+
+
+def _enable(monkeypatch) -> None:
+    monkeypatch.setenv("XQUIK_ENABLED", "true")
+    monkeypatch.setenv("XQUIK_API_KEY", "unit-test-key")
+    monkeypatch.setenv("XQUIK_SEARCH_QUERY", "missile Kyiv")
+
+
+def _tweet(tweet_id: str = "1234567890", username: str = "field_reporter") -> dict:
+    return {
+        "id": tweet_id,
+        "text": "Missile strike reported in Kyiv",
+        "createdAt": "2026-08-22T08:30:00Z",
+        "url": "javascript:alert(1)",
+        "author": {"username": username},
+    }
+
+
+def setup_function() -> None:
+    xquik_news._reset_cache_for_tests()
+
+
+def test_disabled_by_default_never_calls_xquik(monkeypatch) -> None:
+    monkeypatch.delenv("XQUIK_ENABLED", raising=False)
+    monkeypatch.setattr(
+        xquik_news.requests,
+        "get",
+        lambda *args, **kwargs: (_ for _ in ()).throw(AssertionError("unexpected request")),
+    )
+
+    assert xquik_news.fetch_xquik_entries() == []
+
+
+def test_enabled_source_requires_key_and_query(monkeypatch) -> None:
+    monkeypatch.setenv("XQUIK_ENABLED", "true")
+    monkeypatch.delenv("XQUIK_API_KEY", raising=False)
+    monkeypatch.setenv("XQUIK_SEARCH_QUERY", "Kyiv")
+    monkeypatch.setattr(
+        xquik_news.requests,
+        "get",
+        lambda *args, **kwargs: (_ for _ in ()).throw(AssertionError("unexpected request")),
+    )
+
+    assert xquik_news.fetch_xquik_entries() == []
+
+
+def test_request_is_bounded_and_normalizes_untrusted_posts(monkeypatch) -> None:
+    _enable(monkeypatch)
+    monkeypatch.setenv("XQUIK_SEARCH_LIMIT", "999")
+    monkeypatch.setenv("XQUIK_SEARCH_TIMEOUT_S", "999")
+    monkeypatch.setattr(xquik_news, "outbound_user_agent", lambda purpose: "operator-test")
+    seen: dict = {}
+
+    def fake_get(url, *, headers, params, timeout, allow_redirects):
+        seen.update(
+            url=url,
+            headers=headers,
+            params=params,
+            timeout=timeout,
+            allow_redirects=allow_redirects,
+        )
+        return _Response(
+            {
+                "tweets": [
+                    _tweet(),
+                    _tweet(tweet_id="invalid"),
+                    _tweet(tweet_id="9876543210", username="invalid-name"),
+                    {**_tweet(tweet_id="111"), "createdAt": "not-a-date"},
+                    {"id": "42", "text": "", "author": {"username": "empty"}},
+                    "not-an-object",
+                ]
+            }
+        )
+
+    monkeypatch.setattr(xquik_news.requests, "get", fake_get)
+    entries = xquik_news.fetch_xquik_entries()
+
+    assert len(entries) == 1
+    assert entries[0]["link"] == "https://x.com/field_reporter/status/1234567890"
+    assert entries[0]["source"] == "Xquik/@field_reporter"
+    assert entries[0]["published_parsed"].tm_year == 2026
+    assert "javascript:" not in repr(entries)
+    assert seen == {
+        "url": "https://xquik.com/api/v1/x/tweets/search",
+        "headers": {"User-Agent": "operator-test", "x-api-key": "unit-test-key"},
+        "params": {
+            "q": "missile Kyiv",
+            "queryType": "Latest",
+            "limit": 100,
+            "replies": "exclude",
+            "retweets": "exclude",
+        },
+        "timeout": (5, 30),
+        "allow_redirects": False,
+    }
+
+
+def test_successful_results_are_cached_and_copied(monkeypatch) -> None:
+    _enable(monkeypatch)
+    calls = 0
+
+    def fake_get(*args, **kwargs):
+        nonlocal calls
+        calls += 1
+        return _Response({"tweets": [_tweet()]})
+
+    monkeypatch.setattr(xquik_news.requests, "get", fake_get)
+    first = xquik_news.fetch_xquik_entries()
+    first[0]["title"] = "changed"
+    second = xquik_news.fetch_xquik_entries()
+
+    assert calls == 1
+    assert second[0]["title"] == "Missile strike reported in Kyiv"
+
+
+def test_http_failure_keeps_cached_results_without_retry(monkeypatch) -> None:
+    _enable(monkeypatch)
+    responses = [_Response({"tweets": [_tweet()]}), _Response({}, status_code=429)]
+    calls = 0
+
+    def fake_get(*args, **kwargs):
+        nonlocal calls
+        calls += 1
+        return responses.pop(0)
+
+    monkeypatch.setattr(xquik_news.requests, "get", fake_get)
+    clock = iter((1000.0, 2801.0))
+    monkeypatch.setattr(xquik_news.time, "monotonic", lambda: next(clock))
+    expected = xquik_news.fetch_xquik_entries()
+
+    assert xquik_news.fetch_xquik_entries() == expected
+    assert calls == 2
+
+
+def test_failed_query_change_does_not_reuse_another_query(monkeypatch) -> None:
+    _enable(monkeypatch)
+    responses = [_Response({"tweets": [_tweet()]}), _Response({}, status_code=503)]
+    monkeypatch.setattr(
+        xquik_news.requests, "get", lambda *args, **kwargs: responses.pop(0)
+    )
+    assert len(xquik_news.fetch_xquik_entries()) == 1
+
+    monkeypatch.setenv("XQUIK_SEARCH_QUERY", "different region")
+    assert xquik_news.fetch_xquik_entries() == []
+
+
+def test_redirect_and_invalid_json_fail_closed(monkeypatch) -> None:
+    _enable(monkeypatch)
+    responses = [_Response({}, status_code=302), _Response(ValueError("invalid JSON"))]
+
+    monkeypatch.setattr(xquik_news.requests, "get", lambda *args, **kwargs: responses.pop(0))
+    assert xquik_news.fetch_xquik_entries() == []
+    assert xquik_news.fetch_xquik_entries() == []
+
+
+def test_news_fetch_merges_xquik_posts_into_existing_pipeline(monkeypatch) -> None:
+    from services import news_feed_config
+    from services.fetchers import _store, news
+
+    monkeypatch.setenv("NEWS_ENABLED", "true")
+    monkeypatch.setattr(
+        news_feed_config,
+        "get_feeds",
+        lambda: [{"name": "Empty RSS", "url": "https://example.test/rss", "weight": 3}],
+    )
+    monkeypatch.setattr(news, "fetch_with_curl", lambda *args, **kwargs: SimpleNamespace(text=""))
+    monkeypatch.setattr(news.feedparser, "parse", lambda text: SimpleNamespace(entries=[]))
+    monkeypatch.setattr(
+        news,
+        "fetch_xquik_entries",
+        lambda: [
+            {
+                **_tweet(),
+                "title": "Missile strike reported in Kyiv",
+                "summary": "",
+                "link": "https://x.com/field_reporter/status/1234567890",
+                "published": "2026-08-22T08:30:00Z",
+                "published_parsed": time.gmtime(),
+                "source": "Xquik/@field_reporter",
+            }
+        ],
+    )
+    monkeypatch.setattr(news, "enrich_news_items", lambda *args: None)
+    monkeypatch.setattr(news, "detect_breaking_events", lambda *args: None)
+    monkeypatch.setattr(news, "compute_global_threat_level", lambda *args, **kwargs: {"score": 1})
+
+    news.fetch_news()
+
+    item = _store.latest_data["news"][0]
+    assert item["source"] == "Xquik/@field_reporter"
+    assert item["link"] == "https://x.com/field_reporter/status/1234567890"
+    assert item["coords"] == [50.45, 30.523]
+    assert item["risk_score"] == 5
+
+
+def test_news_kill_switch_blocks_xquik_requests(monkeypatch) -> None:
+    from services.fetchers import _store, news
+
+    monkeypatch.setenv("NEWS_ENABLED", "false")
+    monkeypatch.setitem(_store.latest_data, "news", [{"title": "old"}])
+    monkeypatch.setattr(
+        news,
+        "fetch_xquik_entries",
+        lambda: (_ for _ in ()).throw(AssertionError("unexpected Xquik request")),
+    )
+
+    news.fetch_news()
+
+    assert _store.latest_data["news"] == []
+
+
+def test_api_key_is_available_to_the_server_side_settings_registry() -> None:
+    from services.api_settings import ALLOWED_ENV_KEYS, API_REGISTRY
+
+    entry = next(item for item in API_REGISTRY if item["id"] == "xquik_api_key")
+    assert entry["env_key"] == "XQUIK_API_KEY"
+    assert entry["required"] is False
+    assert "XQUIK_API_KEY" in ALLOWED_ENV_KEYS
+
+
+def test_documented_xquik_settings_reach_the_backend_container() -> None:
+    root = Path(__file__).resolve().parents[2]
+    settings = {
+        "XQUIK_ENABLED",
+        "XQUIK_API_KEY",
+        "XQUIK_SEARCH_QUERY",
+        "XQUIK_SEARCH_LIMIT",
+        "XQUIK_SEARCH_INTERVAL_MINUTES",
+        "XQUIK_SEARCH_TIMEOUT_S",
+    }
+    for relative_path in (".env.example", "backend/.env.example", "docker-compose.yml"):
+        text = (root / relative_path).read_text(encoding="utf-8")
+        assert all(setting in text for setting in settings)

docker-compose.yml

@@ -100,6 +100,12 @@ services:
       - FIMI_ENABLED=${FIMI_ENABLED:-false}
       - NUFORC_ENABLED=${NUFORC_ENABLED:-false}
       - NEWS_ENABLED=${NEWS_ENABLED:-true}
+      - XQUIK_ENABLED=${XQUIK_ENABLED:-false}
+      - XQUIK_API_KEY=${XQUIK_API_KEY:-}
+      - XQUIK_SEARCH_QUERY=${XQUIK_SEARCH_QUERY:-}
+      - XQUIK_SEARCH_LIMIT=${XQUIK_SEARCH_LIMIT:-20}
+      - XQUIK_SEARCH_INTERVAL_MINUTES=${XQUIK_SEARCH_INTERVAL_MINUTES:-30}
+      - XQUIK_SEARCH_TIMEOUT_S=${XQUIK_SEARCH_TIMEOUT_S:-10}
       - TELEGRAM_OSINT_ENABLED=${TELEGRAM_OSINT_ENABLED:-true}
       - TELEGRAM_OSINT_CHANNELS=${TELEGRAM_OSINT_CHANNELS:-}
       - TELEGRAM_OSINT_INTERVAL_MINUTES=${TELEGRAM_OSINT_INTERVAL_MINUTES:-60}