diff --git a/.env.example b/.env.example index 341ff54..e02c4f8 100644 --- a/.env.example +++ b/.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 diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index f69f612..bcdc80b 100644 --- a/.github/workflows/ci.yml +++ b/.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 diff --git a/README.md b/README.md index ee49d82..05401b0 100644 --- a/README.md +++ b/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 diff --git a/backend/.env.example b/backend/.env.example index 7ec1449..7eed502 100644 --- a/backend/.env.example +++ b/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. diff --git a/backend/services/api_settings.py b/backend/services/api_settings.py index 2a29e87..e55e1bf 100644 --- a/backend/services/api_settings.py +++ b/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, diff --git a/backend/services/fetchers/news.py b/backend/services/fetchers/news.py index 43d2679..5bb3564 100644 --- a/backend/services/fetchers/news.py +++ b/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") diff --git a/backend/services/fetchers/xquik_news.py b/backend/services/fetchers/xquik_news.py new file mode 100644 index 0000000..e0eb843 --- /dev/null +++ b/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 diff --git a/backend/tests/test_xquik_news.py b/backend/tests/test_xquik_news.py new file mode 100644 index 0000000..49ce597 --- /dev/null +++ b/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) diff --git a/docker-compose.yml b/docker-compose.yml index d6806b3..d91feab 100644 --- a/docker-compose.yml +++ b/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}