diff --git a/answerpath_geo/extractor.py b/answerpath_geo/extractor.py index 8191cd5..890e953 100644 --- a/answerpath_geo/extractor.py +++ b/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 + diff --git a/tests/test_core.py b/tests/test_core.py index 265e7d9..67e02d0 100644 --- a/tests/test_core.py +++ b/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"] +