From c7fd21add3dfc5d0151dc818f389af1760e5b742 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?=E5=B3=AF=E5=B2=B8=E3=80=80=E4=BA=AE?=
 <1920071390@campus.ouj.ac.jp>
Date: Fri, 24 Jul 2026 07:49:35 +0900
Subject: [PATCH] fix(security): reject always-blocked OpenViking endpoints

## Summary
- Normalize OpenViking endpoints through `is_always_blocked_url` and fall back to the default local endpoint when poisoned.
- Keep intentional loopback / LAN self-host working.
- Add focused unit tests.

## Salvage / credit
Memory-provider endpoint floor sibling of RetainDB/Supermemory always-blocked hardening (avoids over-broad #4984-style private-IP bans).

(cherry picked from commit 8fa607d0aedb8c5fca398d7f112b1b25ade54fa2)
---
 plugins/memory/openviking/__init__.py         | 30 +++++++++++++++----
 ...test_openviking_endpoint_always_blocked.py | 18 +++++++++++
 2 files changed, 43 insertions(+), 5 deletions(-)
 create mode 100644 tests/plugins/memory/test_openviking_endpoint_always_blocked.py

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
+    )