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 8fa607d0ae)
This commit is contained in:
峯岸 亮 2026-07-24 07:49:35 +09:00 committed by kshitij
parent ae17163e92
commit c7fd21add3
2 changed files with 43 additions and 5 deletions

View File

@ -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]:

View File

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