diff --git a/plugins/memory/openviking/__init__.py b/plugins/memory/openviking/__init__.py index e59f9467ac236..8736eaac9752e 100644 --- a/plugins/memory/openviking/__init__.py +++ b/plugins/memory/openviking/__init__.py @@ -800,11 +800,31 @@ def _normalize_openviking_url(url: str) -> str: if lower.startswith("::1:"): return f"http://[::1]:{trimmed.rsplit(':', 1)[1]}" if "://" in trimmed: - return trimmed - host, _sep, port = trimmed.partition(":") - if host.lower() in {"localhost", "127.0.0.1"}: - return f"http://{host}:{port or '1933'}" - return trimmed + candidate = trimmed + else: + host, _sep, port = trimmed.partition(":") + if host.lower() in {"localhost", "127.0.0.1"}: + candidate = f"http://{host}:{port or '1933'}" + else: + candidate = trimmed + + # Local / LAN self-host remains allowed; reject cloud-metadata and other + # always-blocked floors so a poisoned endpoint cannot SSRF via memory sync. + try: + from tools.url_safety import is_always_blocked_url + + check_url = candidate if "://" in candidate else f"http://{candidate}" + if is_always_blocked_url(check_url): + logger.warning( + "OpenViking endpoint '%s' targets an always-blocked address; " + "falling back to the default local endpoint.", + candidate, + ) + return _DEFAULT_ENDPOINT + except Exception as exc: + logger.debug("OpenViking always-blocked endpoint check skipped: %s", exc) + + return candidate def _load_profile(path: Path, *, source: str, name: str) -> Optional[_OvcliProfile]: diff --git a/tests/plugins/memory/test_openviking_endpoint_always_blocked.py b/tests/plugins/memory/test_openviking_endpoint_always_blocked.py new file mode 100644 index 0000000000000..271d4c1cc5ed0 --- /dev/null +++ b/tests/plugins/memory/test_openviking_endpoint_always_blocked.py @@ -0,0 +1,18 @@ +"""OpenViking endpoint always-blocked floor.""" + +from plugins.memory.openviking import _DEFAULT_ENDPOINT, _normalize_openviking_url + + +def test_openviking_blocks_metadata_endpoint(): + assert _normalize_openviking_url("http://169.254.169.254/") == _DEFAULT_ENDPOINT + + +def test_openviking_keeps_default_loopback(): + assert _normalize_openviking_url("http://127.0.0.1:1933") == "http://127.0.0.1:1933" + + +def test_openviking_blocks_ecs_metadata_hostname(): + assert ( + _normalize_openviking_url("http://metadata.google.internal/computeMetadata/v1/") + == _DEFAULT_ENDPOINT + )