security: disable implicit remote URL fetching (SSRF v0.2.0)

answerpath_geo/extractor.py

@@ -102,8 +102,12 @@ def handle_data(self, data):
         elif self._in_heading:
             self._curr_heading += data
 
-def extract_website_context(target: str, timeout: int = 4) -> WebsiteContext:
-    """Extracts semantic metadata and brand signals from a website URL or domain."""
+def extract_website_context(target: str, timeout: int = 4, fetch_remote: bool = False) -> WebsiteContext:
+    """Builds safe URL heuristics, optionally fetching remote metadata.
+
+    Remote fetching is opt-in because callers may pass attacker-controlled URLs.
+    The default keeps URL-derived brand/domain context while avoiding SSRF.
+    """
     target = target.strip()
     if not target.startswith(("http://", "https://")):
         url = "https://" + target
@@ -122,6 +126,9 @@ def extract_website_context(target: str, timeout: int = 4) -> WebsiteContext:
         language="fa" if domain_clean.endswith(".ir") else "en"
     )
 
+    if not fetch_remote:
+        return ctx
+
     req = urllib.request.Request(
         url,
         headers={
@@ -234,3 +241,4 @@ def _walk(obj: Any, source: str) -> list[tuple[str, str]]:
         for v in obj:
             out.extend(_walk(v, source))
     return out
+

tests/test_core.py

@@ -86,6 +86,18 @@ def test_website_url_handling_and_mocked_extraction():
     # Injected brand Novin / نوین in generated queries
     assert any("نوین" in q.text or "novin" in q.text.lower() for q in qs)
 
+def test_url_context_does_not_fetch_by_default():
+    with patch("answerpath_geo.extractor.urllib.request.urlopen") as urlopen:
+        ctx = extract_website_context("https://attacker.example/internal")
+    urlopen.assert_not_called()
+    assert ctx.domain == "attacker.example"
+    assert ctx.brand == "Attacker"
+
+def test_remote_fetch_requires_explicit_opt_in():
+    with patch("answerpath_geo.extractor.urllib.request.urlopen") as urlopen:
+        extract_website_context("https://example.com", fetch_remote=True)
+    urlopen.assert_called_once()
+
 def test_benchmark_format_export(tmp_path):
     bench_data = mine("سئو سایت", benchmark_format=True, count=10)
     assert isinstance(bench_data, list)
@@ -105,3 +117,4 @@ def test_benchmark_format_export(tmp_path):
     loaded = json.loads(bench_file.read_text(encoding="utf-8"))
     assert len(loaded) == len(bench_data)
     assert loaded[0]["id"] == item["id"]
+