kenzuya/unshackle
1
0
Fork 2
mirror of https://github.com/unshackle-dl/unshackle.git synced 2026-11-11 19:45:06 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(tunnel): block local networks instead of allowing public ones for --local-proxy

Browse Source 
A VPN's synthetic DNS answer (fake-IP 198.18.0.0/15, 192.0.0.69) now works for a host name. An IP from the server, in any numeric form, must be public; dials to the client itself are closed.
This commit is contained in:
imSp4rky
2026-10-27 10:07:23 -06:00
parent 06f87dafa3
commit 8454f40929
3 changed files with 176 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/guide/proxies-and-vpn.md
+11 -8
View File
@@ -315,14 +315,17 @@ machine opens none, because the client opens the connection to the server. A NAT
firewall, or a Cloudflare tunnel in front of the server does not stop it.
The service traffic stays encrypted from the server to the service, so the client relays only
encrypted bytes. The client refuses to connect to `localhost`, and to a host name or IP address
that resolves only to a loopback, private or link-local address, so the server cannot reach
your local network.
A VPN or proxy app in TUN mode with fake-IP DNS answers a host name with an address from
`198.18.0.0/15` or `2001:2::/48` and maps it back to the real host. The client accepts that
address for a host name, but not as an IP address from the server. An app set to a different
fake-IP range does not work with `--local-proxy`. With fake-IP DNS, the client cannot see the
real address of a host name, so it cannot stop a host name that points into your local network.
encrypted bytes. The client refuses to connect to `localhost`, and to a host name that resolves
only to an address on your machine or local network: loopback, private, link-local, CGNAT
(`100.64.0.0/10`) or multicast. An IP address that the server sends must also be public, and
the client closes a connection that reaches one of its own addresses. Other devices on your
network that have a public IPv6 address, and your router's public address, are not blocked.
Some VPN and proxy apps answer a host name with a synthetic address, for example fake-IP DNS in
TUN mode (`198.18.0.0/15`), and map it back to the real host. The client accepts any such
address for a host name, so `--local-proxy` works with these apps. The client cannot see the
real address behind a synthetic one, so it cannot stop a host name that points into your local
network.
!!! note "Each request crosses the tunnel twice"
A request from the server goes to your machine and then to the service, so it takes
tests/remote/unit/test_tunnel.py
+96 -16
View File
@@ -28,7 +28,8 @@ from unshackle.core.tunnel import (
TunnelClient,
connect_public,
frame,
is_fake_ip,
is_ip_literal,
is_local_ip,
is_public_ip,
tunnel_proxy,
)
@@ -78,6 +79,7 @@ def test_requests_through_the_tunnel_reach_the_client_side_target(
server: str, target: int, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setattr(tunnel, "is_public_ip", lambda address: True) # the test target is on loopback
monkeypatch.setattr(tunnel, "dialled_itself", lambda writer: False)
client = TunnelClient(server, {"X-Secret-Key": "alice"})
tunnel_id = client.start()
@@ -114,6 +116,7 @@ def test_the_client_refuses_a_target_on_its_own_network(server: str, target: int
("10.0.0.1", False),
("::1", False),
("198.18.2.36", False),
("64:ff9b::c0a8:101", False),
],
)
def test_is_public_ip(address: str, public: bool) -> None:
@@ -121,35 +124,99 @@ def test_is_public_ip(address: str, public: bool) -> None:
@pytest.mark.parametrize(
("address", "fake"),
("address", "local"),
[
("198.18.2.36", True),
("198.19.255.1", True),
("198.20.0.1", False),
("::ffff:198.18.0.1", False),
("2001:2::1", True),
("2001:2:1::1", False),
("fdfe:dcba:9876::1", False),
("8.8.8.8", False),
("198.18.2.36", False),
("2001:2::1", False),
("192.0.0.69", False),
("192.0.0.88", False),
("127.0.0.1", True),
("10.0.0.1", True),
("172.16.0.1", True),
("192.168.1.1", True),
("169.254.169.254", True),
("100.64.0.1", True),
("192.0.0.1", True),
("0.0.0.0", True),
("224.0.0.1", True),
("::1", True),
("fdfe:dcba:9876::1", True),
("fe80::1%eth0", True),
("::ffff:192.168.1.1", True),
("64:ff9b::c0a8:101", True),
("64:ff9b::808:808", False),
("::", True),
("::7f00:1", True),
("::ffff:0:7f00:1", True),
("64:ff9b:1::c0a8:101", True),
("192.0.0.9", True),
("192.0.0.10", True),
("not-an-address", True),
],
)
def test_is_fake_ip(address: str, fake: bool) -> None:
assert is_fake_ip(address) is fake
def test_is_local_ip(address: str, local: bool) -> None:
assert is_local_ip(address) is local
def test_a_name_that_resolves_to_a_fake_ip_is_dialled(monkeypatch: pytest.MonkeyPatch) -> None:
@pytest.mark.parametrize("address", ["198.18.2.36", "192.0.0.69", "2001:2::1", "8.8.8.8"])
def test_a_name_that_resolves_to_a_non_local_address_is_dialled(monkeypatch: pytest.MonkeyPatch, address: str) -> None:
async def resolve(self: object, host: str, port: int, **_: object) -> list:
return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("198.18.2.36", port))]
return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", (address, port))]
dialled: list[str] = []
async def open_connection(address: str, port: int) -> tuple:
dialled.append(address)
return None, None
return None, FakeWriter(("10.0.0.2", 50000), (address, port))
monkeypatch.setattr(asyncio.BaseEventLoop, "getaddrinfo", resolve)
monkeypatch.setattr(asyncio, "open_connection", open_connection)
asyncio.run(connect_public("api.example", 443))
assert dialled == ["198.18.2.36"]
assert dialled == [address]
class FakeWriter:
def __init__(self, sockname: tuple, peername: tuple) -> None:
self.extra = {"sockname": sockname, "peername": peername}
self.closed = False
def get_extra_info(self, name: str) -> tuple:
return self.extra[name]
def close(self) -> None:
self.closed = True
@pytest.mark.parametrize(
("host", "literal"),
[
("8.8.8.8", True),
("::1", True),
("198.18.1", True),
("3323068417", True),
("0x7f.1", True),
("api.example", False),
],
)
def test_is_ip_literal(host: str, literal: bool) -> None:
assert is_ip_literal(host) is literal
def test_a_dial_that_reaches_the_client_itself_is_refused(monkeypatch: pytest.MonkeyPatch) -> None:
async def resolve(self: object, host: str, port: int, **_: object) -> list:
return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("198.18.0.1", port))]
writer = FakeWriter(("198.18.0.1", 50000), ("198.18.0.1", 443))
async def open_connection(address: str, port: int) -> tuple:
return None, writer
monkeypatch.setattr(asyncio.BaseEventLoop, "getaddrinfo", resolve)
monkeypatch.setattr(asyncio, "open_connection", open_connection)
with pytest.raises(PermissionError, match="this machine"):
asyncio.run(connect_public("tun.example", 443))
assert writer.closed
def test_localhost_is_refused() -> None:
@@ -157,7 +224,18 @@ def test_localhost_is_refused() -> None:
asyncio.run(connect_public("localhost", 443))
@pytest.mark.parametrize(("host", "address"), [("rebind.example", "192.168.1.1"), ("198.18.0.1", "198.18.0.1")])
@pytest.mark.parametrize(
("host", "address"),
[
("rebind.example", "192.168.1.1"),
("metadata.example", "169.254.169.254"),
("nat64.example", "64:ff9b::c0a8:101"),
("198.18.0.1", "198.18.0.1"),
("192.0.0.69", "192.0.0.69"),
("198.18.1", "198.18.0.1"),
("3221225541", "192.0.0.69"),
],
)
def test_a_target_that_resolves_to_a_local_address_is_refused(
monkeypatch: pytest.MonkeyPatch, host: str, address: str
) -> None:
@@ -202,6 +280,7 @@ def test_the_tunnel_port_refuses_a_proxy_uri_without_the_tunnel_credentials(
server: str, target: int, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setattr(tunnel, "is_public_ip", lambda address: True)
monkeypatch.setattr(tunnel, "dialled_itself", lambda writer: False)
client = TunnelClient(server, {"X-Secret-Key": "alice"})
tunnel_id = client.start()
port = TUNNELS[tunnel_id].port
@@ -371,6 +450,7 @@ def test_a_dial_the_server_gave_up_on_is_dropped_by_the_client(
) -> None:
"""When the server times out an OPEN, the client must not keep the connection it makes later."""
monkeypatch.setattr(tunnel, "is_public_ip", lambda address: True)
monkeypatch.setattr(tunnel, "dialled_itself", lambda writer: False)
monkeypatch.setattr(tunnel, "OPEN_TIMEOUT", 0.3)
real_connect = tunnel.connect_public
unshackle/core/tunnel.py
+69 -13
View File
@@ -30,7 +30,30 @@ HEARTBEAT = 20
GRACE = 60
RELINK_WAIT = 10
OPEN_TIMEOUT = 15
FAKE_IP_POOLS = (ipaddress.ip_network("198.18.0.0/15"), ipaddress.ip_network("2001:2::/48"))
LOCAL_NETWORKS = tuple(
ipaddress.ip_network(n)
for n in (
"0.0.0.0/8",
"10.0.0.0/8",
"100.64.0.0/10", # CGNAT, also tailnets
"127.0.0.0/8",
"169.254.0.0/16",
"172.16.0.0/12",
"192.0.0.0/29", # DS-Lite endpoints on the user's router
"192.0.0.9/32", # port-control and TURN anycast, answered by the user's ISP or router
"192.0.0.10/32",
"192.168.0.0/16",
"224.0.0.0/4",
"255.255.255.255/32",
"::/96", # also IPv4-compatible addresses, which some stacks route to the embedded IPv4
"::ffff:0:0:0/96", # IPv4-translated
"fc00::/7",
"fe80::/10",
"fec0::/10",
"ff00::/8",
)
)
NAT64 = (ipaddress.ip_network("64:ff9b::/96"), ipaddress.ip_network("64:ff9b:1::/48"))
MAX_TUNNELS = 4
MAX_STREAMS = 256
CHUNK = 65536
@@ -290,31 +313,64 @@ def parse_ip(address: str) -> Optional[ipaddress.IPv4Address | ipaddress.IPv6Add
return None
def is_local_ip(address: str) -> bool:
"""True for an address on the client's own machine or LAN, or one that is not an address. An IPv4 address
inside an IPv4-mapped or NAT64 address counts as itself."""
ip = parse_ip(address)
if isinstance(ip, ipaddress.IPv6Address):
if ip.ipv4_mapped:
ip = ip.ipv4_mapped
elif any(ip in prefix for prefix in NAT64):
ip = ipaddress.IPv4Address(int(ip) & 0xFFFFFFFF)
return ip is None or any(ip in network for network in LOCAL_NETWORKS)
def is_public_ip(address: str) -> bool:
return (ip := parse_ip(address)) is not None and ip.is_global
return (ip := parse_ip(address)) is not None and ip.is_global and not is_local_ip(address)
def is_fake_ip(address: str) -> bool:
"""True for an address from a pool that fake-IP DNS in a TUN-mode VPN or proxy hands out for a name."""
return (ip := parse_ip(address)) is not None and any(ip in pool for pool in FAKE_IP_POOLS)
def is_ip_literal(host: str) -> bool:
"""True when the resolver reads ``host`` as a number, which includes short forms such as ``10.1``."""
if parse_ip(host):
return True
try:
socket.inet_aton(host)
except (OSError, ValueError):
return False
return True
def dialled_itself(writer: asyncio.StreamWriter) -> bool:
"""True when the connection runs from one of the client's own addresses to the same address."""
return parse_ip(str(writer.get_extra_info("sockname")[0])) == parse_ip(str(writer.get_extra_info("peername")[0]))
async def connect_public(host: str, port: int) -> tuple[asyncio.StreamReader, asyncio.StreamWriter]:
"""Connect to the first public address of ``host``. The server picks the targets, so a name that resolves
only to the client's own machine or LAN is refused. A fake IP passes only when it comes from a name lookup:
the TUN maps it back to the real host, which the client cannot check."""
"""Connect to the first allowed address of ``host``. The server picks the targets, so the client never dials
its own machine or LAN. A name may resolve to any other address, because a VPN app's DNS can answer with a
synthetic one (such as fake-IP DNS in a TUN) and map it back to the real host, which the client cannot
check. An IP address from the server must be public."""
infos = await asyncio.get_running_loop().getaddrinfo(host, port, type=socket.SOCK_STREAM)
resolved = [str(info[4][0]) for info in infos]
from_name = parse_ip(host) is None
addresses = [a for a in resolved if is_public_ip(a) or (from_name and is_fake_ip(a))]
if not addresses:
raise PermissionError(f"{host} resolves only to local addresses ({', '.join(resolved)})")
if is_ip_literal(host):
addresses = [a for a in resolved if is_public_ip(a)]
if not addresses:
raise PermissionError(f"{host} is not a public address")
else:
addresses = [a for a in resolved if not is_local_ip(a)]
if not addresses:
raise PermissionError(f"{host} resolves only to local addresses ({', '.join(resolved)})")
error: OSError = OSError(f"could not connect to {host}")
for address in addresses:
try:
return await asyncio.wait_for(asyncio.open_connection(address, port), 10)
reader, writer = await asyncio.wait_for(asyncio.open_connection(address, port), 10)
except (OSError, asyncio.TimeoutError) as e:
error = OSError(str(e))
continue
if not dialled_itself(writer):
return reader, writer
writer.close()
error = PermissionError(f"{host} resolves to this machine ({address})")
raise error