jetty防止Dos攻击的filter实现分析

jetty的org.eclipse.jetty.servlets.DoSFilter类是用来实现Dos攻击预防的filter，里面涉及到一些变量，先了解下变量的含义：

 

    protected long _delayMs;超过最大处理请求数当前请求的等待时间，-1立即拒绝，0，无限等待，正数表达等待的毫秒数

    protected long _throttleMs;异步等待获取信号量的时间

    protected long _maxWaitMs;阻塞等待获取信号量的时间

    protected long _maxRequestMs;请求处理最大时间限制

    protected long _maxIdleTrackerMs;跟踪连接是否断开的最大等待时间

    protected int _throttledRequests;允许在等待队列中等待获取信号量的请求数

    protected int _maxRequestsPerSec; 每秒允许处理最多的请求数，超过将延迟，异步等待。

 

    protected boolean _insertHeaders; 是否往response写入dosfilter信息，默认true

    protected boolean _trackSessions;是否根据session来检测dos攻击，默认true

    protected boolean _remotePort;是否根据ip+port来检测dos攻击，默认false

protected String _whitelistStr;  白名单 ip白名单列表，这些通过都通过配置servlet的init-p

 

aram可以来重新设置。

 

首先看看init方法的初始化设置：

 

 public void init(FilterConfig filterConfig)
    {
        _context = filterConfig.getServletContext();

        _queue = new Queue[getMaxPriority() + 1];
        _listener = new ContinuationListener[getMaxPriority() + 1];
        for (int p = 0; p < _queue.length; p++)
        {
            _queue[p] = new ConcurrentLinkedQueue<Continuation>();

            final int priority=p;
            _listener[p] = new ContinuationListener()
            {
                public void onComplete(Continuation continuation)
                {
                }

                public void onTimeout(Continuation continuation)
                {
                    _queue[priority].remove(continuation);
                }
            };
        }

        _rateTrackers.clear();

        int baseRateLimit = __DEFAULT_MAX_REQUESTS_PER_SEC;
        if (filterConfig.getInitParameter(MAX_REQUESTS_PER_S_INIT_PARAM) != null)
            baseRateLimit = Integer.parseInt(filterConfig.getInitParameter(MAX_REQUESTS_PER_S_INIT_PARAM));
        _maxRequestsPerSec = baseRateLimit;

        long delay = __DEFAULT_DELAY_MS;
        if (filterConfig.getInitParameter(DELAY_MS_INIT_PARAM) != null)
            delay = Integer.parseInt(filterConfig.getInitParameter(DELAY_MS_INIT_PARAM));
        _delayMs = delay;

        int throttledRequests = __DEFAULT_THROTTLE;
        if (filterConfig.getInitParameter(THROTTLED_REQUESTS_INIT_PARAM) != null)
            throttledRequests = Integer.parseInt(filterConfig.getInitParameter(THROTTLED_REQUESTS_INIT_PARAM));
        _passes = new Semaphore(throttledRequests,true);
        _throttledRequests = throttledRequests;

        long wait = __DEFAULT_WAIT_MS;
        if (filterConfig.getInitParameter(MAX_WAIT_INIT_PARAM) != null)
            wait = Integer.parseInt(filterConfig.getInitParameter(MAX_WAIT_INIT_PARAM));
        _maxWaitMs = wait;

        long suspend = __DEFAULT_THROTTLE_MS;
        if (filterConfig.getInitParameter(THROTTLE_MS_INIT_PARAM) != null)
            suspend = Integer.parseInt(filterConfig.getInitParameter(THROTTLE_MS_INIT_PARAM));
        _throttleMs = suspend;

        long maxRequestMs = __DEFAULT_MAX_REQUEST_MS_INIT_PARAM;
        if (filterConfig.getInitParameter(MAX_REQUEST_MS_INIT_PARAM) != null )
            maxRequestMs = Long.parseLong(filterConfig.getInitParameter(MAX_REQUEST_MS_INIT_PARAM));
        _maxRequestMs = maxRequestMs;

        long maxIdleTrackerMs = __DEFAULT_MAX_IDLE_TRACKER_MS_INIT_PARAM;
        if (filterConfig.getInitParameter(MAX_IDLE_TRACKER_MS_INIT_PARAM) != null )
            maxIdleTrackerMs = Long.parseLong(filterConfig.getInitParameter(MAX_IDLE_TRACKER_MS_INIT_PARAM));
        _maxIdleTrackerMs = maxIdleTrackerMs;

        _whitelistStr = "";
        if (filterConfig.getInitParameter(IP_WHITELIST_INIT_PARAM) !=null )
            _whitelistStr = filterConfig.getInitParameter(IP_WHITELIST_INIT_PARAM);
        initWhitelist();

        String tmp = filterConfig.getInitParameter(INSERT_HEADERS_INIT_PARAM);
        _insertHeaders = tmp==null || Boolean.parseBoolean(tmp);

        tmp = filterConfig.getInitParameter(TRACK_SESSIONS_INIT_PARAM);
        _trackSessions = tmp==null || Boolean.parseBoolean(tmp);

        tmp = filterConfig.getInitParameter(REMOTE_PORT_INIT_PARAM);
        _remotePort = tmp!=null&& Boolean.parseBoolean(tmp);

        _requestTimeoutQ.setNow();
        _requestTimeoutQ.setDuration(_maxRequestMs);

        _trackerTimeoutQ.setNow();
        _trackerTimeoutQ.setDuration(_maxIdleTrackerMs);

        _running=true;
        _timerThread = (new Thread()
        {
            public void run()
            {
                try
                {
                    while (_running)
                    {
                        long now;
                        synchronized (_requestTimeoutQ)
                        {
                            now = _requestTimeoutQ.setNow();
                            _requestTimeoutQ.tick();
                        }
                        synchronized (_trackerTimeoutQ)
                        {
                            _trackerTimeoutQ.setNow(now);
                            _trackerTimeoutQ.tick();
                        }
                        try
                        {
                            Thread.sleep(100);
                        }
                        catch (InterruptedException e)
                        {
                            Log.ignore(e);
                        }
                    }
                }
                finally
                {
                    Log.info("DoSFilter timer exited");
                }
            }
        });
        _timerThread.start();

        if (_context!=null && Boolean.parseBoolean(filterConfig.getInitParameter(MANAGED_ATTR_INIT_PARAM)))
            _context.setAttribute(filterConfig.getFilterName(),this);
    }

 _queue是用来保存当前当前通过header，session，ip等检测类型的请求队列，然后就是一堆参数的初始化设置，最后

又有两个queue _requestTimeoutQ是来保存每个请求的处理时间超时检测的队列；_trackerTimeoutQ是来检测请求对应连接是否已经关闭超时的检测队列；最后启动一个_timerThread线程来进行这两个队列的超时检测。

 

然后再看下dofilter的处理

  public void doFilter(ServletRequest request, ServletResponse response, FilterChain filterchain) throws IOException, ServletException
    {
        final HttpServletRequest srequest = (HttpServletRequest)request;
        final HttpServletResponse sresponse = (HttpServletResponse)response;

        final long now=_requestTimeoutQ.getNow();

        // Look for the rate tracker for this request
        RateTracker tracker = (RateTracker)request.getAttribute(__TRACKER); 

        if (tracker==null) //如果request没有进行过dosfilter的处理
        {
            // This is the first time we have seen this request.

            // get a rate tracker associated with this request, and record one hit
            tracker = getRateTracker(request); //根据request生成dos跟踪的对象，看下面方法

            // Calculate the rate and check it is over the allowed limit
            final boolean overRateLimit = tracker.isRateExceeded(now);//判断是否已经超过每秒最大处理请求数

            // pass it through if  we are not currently over the rate limit
            if (!overRateLimit)//如果没有超过，则正常处理
            {
                doFilterChain(filterchain,srequest,sresponse);
                return;
            }

            // We are over the limit.
            Log.warn("DOS ALERT: ip="+srequest.getRemoteAddr()+",session="+srequest.getRequestedSessionId()+",user="+srequest.getUserPrincipal());

            // So either reject it, delay it or throttle it
            switch((int)_delayMs) //根据当前配置的延时时间
            {
                case -1: //如果为-1，则直接拒绝
                {
                    // Reject this request
                    if (_insertHeaders) //是否把dosfilter处理插入response的header
                        ((HttpServletResponse)response).addHeader("DoSFilter","unavailable");
                    ((HttpServletResponse)response).sendError(HttpServletResponse.SC_SERVICE_UNAVAILABLE);
                    return;
                }
                case 0: //如果为0，则当前线程继续等待处理，
                {
                    // fall through to throttle code
                    request.setAttribute(__TRACKER,tracker);
                    break;
                }
                default://其他字段，则设置request的timeout时间为_delayMs，并且挂起当前线程，返回
                {
                    // insert a delay before throttling the request
                    if (_insertHeaders)
                        ((HttpServletResponse)response).addHeader("DoSFilter","delayed");
                    Continuation continuation = ContinuationSupport.getContinuation(request);
                    request.setAttribute(__TRACKER,tracker);
                    if (_delayMs > 0)
                        continuation.setTimeout(_delayMs);
                    continuation.suspend();
                    return;
                }
            }
        }

//_delayMs为0是，当前请求继续等待
        // Throttle the request
        boolean accepted = false;
        try
        {
            // check if we can afford to accept another request at this time
            accepted = _passes.tryAcquire(_maxWaitMs,TimeUnit.MILLISECONDS);//判断当前处理请求队列是否已经有处理完的请求，处理完的会释放信号量，则当前请求线程可以获取信号量

            if (!accepted) //如果不能获取，则把当前请求设置为异步等待，异步等待的时间为_throttleMs，
            {
                // we were not accepted, so either we suspend to wait,or if we were woken up we insist or we fail
                final Continuation continuation = ContinuationSupport.getContinuation(request);

                Boolean throttled = (Boolean)request.getAttribute(__THROTTLED);
                if (throttled!=Boolean.TRUE && _throttleMs>0)
                {
                    int priority = getPriority(request,tracker);
                    request.setAttribute(__THROTTLED,Boolean.TRUE);
                    if (_insertHeaders)
                        ((HttpServletResponse)response).addHeader("DoSFilter","throttled");
                    if (_throttleMs > 0)
                        continuation.setTimeout(_throttleMs);
                    continuation.suspend();

                    continuation.addContinuationListener(_listener[priority]);
                    _queue[priority].add(continuation);
                    return;
                }
                // else were we resumed?
                else if  (request.getAttribute("javax.servlet.resumed")==Boolean.TRUE) 
                {//如果线程中心被唤醒，则可以获取到信号量，
                    // we were resumed and somebody stole our pass, so we wait for the next one.
                    _passes.acquire();
                    accepted = true;
                }
            }

            // if we were accepted (either immediately or after throttle)
            if (accepted) //获取到了，继续执行，
                // call the chain
                doFilterChain(filterchain,srequest,sresponse);
            else
            {
                // fail the request
                if (_insertHeaders)
                    ((HttpServletResponse)response).addHeader("DoSFilter","unavailable");
                ((HttpServletResponse)response).sendError(HttpServletResponse.SC_SERVICE_UNAVAILABLE);
            }
        }
        catch (InterruptedException e)
        {
            _context.log("DoS",e);
            ((HttpServletResponse)response).sendError(HttpServletResponse.SC_SERVICE_UNAVAILABLE);
        }
        finally
        {
            if (accepted) //执行完了之后释放当前的信号量，唤醒等待队列中的第一个请求进行处理
            {
                // wake up the next highest priority request.
                for (int p = _queue.length; p-- > 0;)
                {
                    Continuation continuation = _queue[p].poll();
                    if (continuation != null && continuation.isSuspended())
                    {
                        continuation.resume();
                        break;
                    }
                }
                _passes.release();
            }
        }
    }

  public RateTracker getRateTracker(ServletRequest request)
    {
        HttpServletRequest srequest = (HttpServletRequest)request;
        HttpSession session=srequest.getSession(false);

        String loadId = extractUserId(request);
        final int type;
        if (loadId != null)
        {
            type = USER_AUTH;
        }
        else
        {
            if (_trackSessions && session!=null && !session.isNew())
            {
                loadId=session.getId();
                type = USER_SESSION;
            }
            else
            {
                loadId = _remotePort?(request.getRemoteAddr()+request.getRemotePort()):request.getRemoteAddr();
                type = USER_IP;
            }
        }

        RateTracker tracker=_rateTrackers.get(loadId);

        if (tracker==null)
        {
            RateTracker t;
            if (_whitelist.contains(request.getRemoteAddr())) //如果在白名单中，则isRateExceeded一直返回false
            {
                t = new FixedRateTracker(loadId,type,_maxRequestsPerSec);
            }
            else
            {
                t = new RateTracker(loadId,type,_maxRequestsPerSec);
            }

            tracker=_rateTrackers.putIfAbsent(loadId,t);
            if (tracker==null)
                tracker=t;

            if (type == USER_IP)
            {
                // USER_IP expiration from _rateTrackers is handled by the _trackerTimeoutQ
                synchronized (_trackerTimeoutQ)
                {
                    _trackerTimeoutQ.schedule(tracker);
                }
            }
            else if (session!=null)
                // USER_SESSION expiration from _rateTrackers are handled by the HttpSessionBindingListener
                session.setAttribute(__TRACKER,tracker);
        }

        return tracker;
    }