packetprocessor.cc

网络流量采集及分析软件

/*!\file   PacketProcessor.cc

    Copyright 2003-2004 Fraunhofer Institute for Open Communication Systems (FOKUS),
                        Berlin, Germany

    This file is part of Network Measurement and Accounting System (NETMATE).

    NETMATE is free software; you can redistribute it and/or modify 
    it under the terms of the GNU General Public License as published by 
    the Free Software Foundation; either version 2 of the License, or
    (at your option) any later version.

    NETMATE is distributed in the hope that it will be useful, 
    but WITHOUT ANY WARRANTY; without even the implied warranty of 
    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
    GNU General Public License for more details.

    You should have received a copy of the GNU General Public License
    along with this software; if not, write to the Free Software 
    Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA

    Description:
    manages and applies packet processing modules

    $Id: PacketProcessor.cc,v 1.6 2005/12/07 08:10:56 s_zander Exp $
*/

#include "PacketProcessor.h"
#include "Module.h"
#include "ParserFcts.h"


//!\short  default number of packet buffers to reserve if none is configured in the meter config file (item: PacketQueueBuffers)
static const int  DEF_PACKET_BUFFERS = 2000;


/* ------------------------- PacketProcessor ------------------------- */

PacketProcessor::PacketProcessor(ConfigManager *cnf, int threaded, string moduleDir ) 
    : MeterComponent(cnf, "PacketProcessor", threaded),
      numRules(0), expt(NULL)
{
    string txt;
    
#ifdef DEBUG
    log->dlog(ch,"Starting");
#endif

    if (moduleDir.empty()) {
        if ((txt = cnf->getValue("ModuleDir", "PKTPROCESSOR")) != "") {
            moduleDir = txt;
        }
    }

    if ((txt = cnf->getValue("PacketQueueBuffers",  "PKTPROCESSOR")) != "") {
        queue = new PacketQueue(ParserFcts::parseULong(txt, 0), threaded);
    } else {
        queue = new PacketQueue(DEF_PACKET_BUFFERS, threaded);
    }

    try {
        loader = new ModuleLoader(cnf, moduleDir.c_str() /*module (lib) basedir*/,
                                  cnf->getValue("Modules", "PKTPROCESSOR"),/*modlist*/
                                  "Proc" /*channel name prefix*/);
    } catch (Error &e) {
        saveDelete(queue);
        throw e;
    }
}


/* ------------------------- ~PacketProcessor ------------------------- */

PacketProcessor::~PacketProcessor()
{

#ifdef DEBUG
    log->dlog(ch,"Shutdown");
#endif

#ifdef ENABLE_THREADS
    if (threaded) {
        mutexLock(&maccess);
        stop();
        mutexUnlock(&maccess);
        mutexDestroy(&maccess);
    }
#endif

    // destroyFlowRecord for all rules
    for (ruleActionListIter_t r = rules.begin(); r != rules.end(); r++) {
        for (ppactionListIter_t i = r->actions.begin(); 
             i != r->actions.end(); i++) {
            if (i->flowData != NULL) {
                i->mapi->destroyFlowRec(i->flowData);
            }
            saveDeleteArr(i->params);
        }
        
        if (r->flowKeyLen > 0) {
            saveDeleteArr(r->flowKeyList);
        }

        if (r->auto_flows) {
            for (flowListIter_t i = r->flows->getFlows()->begin(); 
                 i != r->flows->getFlows()->end(); ++i) {
                for (ppactionListIter_t j = i->second.actions.begin(); j != i->second.actions.end(); j++) {
                    j->mapi->destroyFlowRec(j->flowData);
                    j->flowData = NULL; 
                    j->params = NULL;
                    
                    // FIXME disable timers
                }
            }

            saveDelete(r->flows);
        }
    }

    // discard the Module Loader
    saveDelete(loader);

    // destroy the packet queue
    saveDelete(queue);
}


// check a ruleset (the filter part)
void PacketProcessor::checkRules(ruleDB_t *rules)
{
    ruleDBIter_t iter;
    
    for (iter = rules->begin(); iter != rules->end(); iter++) {
        checkRule(*iter);
    }
}


// add rules
void PacketProcessor::addRules( ruleDB_t *rules, EventScheduler *e )
{
    ruleDBIter_t iter;
   
    for (iter = rules->begin(); iter != rules->end(); iter++) {
        addRule(*iter, e);
    }
}


// delete rules
void PacketProcessor::delRules(ruleDB_t *rules)
{
    ruleDBIter_t iter;

    for (iter = rules->begin(); iter != rules->end(); iter++) {
        delRule(*iter);
    }
}


int PacketProcessor::checkRule(Rule *r)
{
    int ruleId;
    actionList_t *actions;
    ppaction_t a;

    ruleId  = r->getUId();
    actions = r->getActions();

#ifdef DEBUG
    log->dlog(ch, "checking Rule %s.%s", r->getSetName().c_str(), r->getRuleName().c_str());
#endif  

    try {
        AUTOLOCK(threaded, &maccess);

        for (actionListIter_t iter = actions->begin(); iter != actions->end(); iter++) {
            Module *mod;
            string mname = iter->name;
       
            a.flowData = NULL;
            a.module = NULL;
            a.params = NULL;

            // load Action Module used by this rule
            mod = loader->getModule(mname.c_str());
            a.module = dynamic_cast<ProcModule*> (mod);

            if (a.module != NULL) { // is it a processing kind of module

                a.mapi = a.module->getAPI();

                // init module
                a.params = ConfigManager::getParamList(iter->conf);
                int ret = (a.mapi)->initFlowRec(a.params, &a.flowData);

                if (ret < 0) {
                    throw Error("Invalid parameters for module %s", mname.c_str());
                }

                saveDeleteArr(a.params);
                a.params = NULL;

                // free memory
                (a.mapi)->destroyFlowRec(a.flowData);
                a.flowData = NULL;

                //release packet processing modules already loaded for this rule
                loader->releaseModule(a.module);
                a.module = NULL;
            }
        }
    	
    } catch (Error &e) { 
        log->elog(ch, e);

        if (a.params != NULL) {
            saveDeleteArr(a.params);
        }

        // free memory
        if (a.flowData != NULL) {
            (a.mapi)->destroyFlowRec(a.flowData);
        }
            
        //release packet processing modules already loaded for this rule
        if (a.module) {
            loader->releaseModule(a.module);
        }
       
        throw e;
    }
    return 0;
}


/* ------------------------- addRule ------------------------- */

int PacketProcessor::addRule( Rule *r, EventScheduler *e )
{
    int ruleId;
    ruleActions_t entry;
    actionList_t *actions;

    ruleId  = r->getUId();
    actions = r->getActions();

#ifdef DEBUG
    log->dlog(ch, "adding Rule #%d", ruleId);
#endif  

    AUTOLOCK(threaded, &maccess);  

    entry.lastPkt = 0;
    entry.packets = 0;
    entry.bytes = 0;
    entry.flowKeyLen = 0;
    entry.flowKeyList = r->getFlowKeyList();
    entry.flist = r->getFilter();
    entry.auto_flows = r->isFlagEnabled(RULE_AUTO_FLOWS);
    entry.bidir = r->isBidir();
    entry.seppaths = r->sepPaths();
    entry.newFlow = 1;
    entry.rule = r;
    entry.flows = NULL;
    if (entry.auto_flows) {
        entry.flows = new FlowCreator();
    }

    try {

        int cnt = 0;
        for (actionListIter_t iter = actions->begin(); iter != actions->end(); iter++) {
            ppaction_t a;
            Module *mod;
            string mname = iter->name;
            	    
            // load Action Module used by this rule
            mod = loader->getModule(mname.c_str());
            a.module = dynamic_cast<ProcModule*> (mod);

            if (a.module != NULL) { // is it a processing kind of module

                a.mapi = a.module->getAPI();

                // init module
                a.params = ConfigManager::getParamList(iter->conf);
                a.flowData = NULL;
                int ret = (a.mapi)->initFlowRec(a.params, &a.flowData);

                // if packet proc modules requires bidir matching
                // then set rule to bidir
                // FIXME not a well defined method...
                if (ret == 1) {
                    r->setBidir();
                }

                // init timers
                addTimerEvents(ruleId, cnt, a, *e);
	
                entry.actions.push_back(a);
            }

            cnt++;
        }
    
        // make sure the vector of rules is large enough
        if ((unsigned int)ruleId + 1 > rules.size()) {
            rules.reserve(ruleId*2 + 1);
            rules.resize(ruleId + 1 );
        }
        // success ->enter struct into internal table
        rules[ruleId] = entry;
	
    } catch (Error &e) { 
        log->elog(ch, e);
	
        for (ppactionListIter_t i = entry.actions.begin(); 
             i != entry.actions.end(); i++) {

            saveDelete(i->params);

            (i->mapi)->destroyFlowRec(i->flowData);
	    
            //release packet processing modules already loaded for this rule
            if (i->module) {
                loader->releaseModule(i->module);
            }
        }
        // empty the list itself
        entry.actions.clear();

        throw e;
    }
    return 0;
}


/* ------------------------- delRule ------------------------- */

int PacketProcessor::delRule( Rule *r )
{
    ruleActions_t *ra;
    int ruleId = r->getUId();

#ifdef DEBUG
    log->dlog(ch, "deleting Rule #%d", ruleId);
#endif

    AUTOLOCK(threaded, &maccess);

    ra = &rules[ruleId];

    if (ra->auto_flows) {
        for (flowListIter_t i = ra->flows->getFlows()->begin(); 
             i != ra->flows->getFlows()->end(); ++i) {
            for (ppactionListIter_t j = i->second.actions.begin(); j != i->second.actions.end(); j++) {
                j->mapi->destroyFlowRec(j->flowData);
                j->flowData = NULL; 
                j->params = NULL;

                // FIXME disable timers
            }
        }

        saveDelete(ra->flows);
    }


    // now free flow data and release used Modules
    for (ppactionListIter_t i = ra->actions.begin(); i != ra->actions.end(); i++) {

        // dismantle flow data structure with module function
        i->mapi->destroyFlowRec(i->flowData);
        i->flowData = NULL;
        
        if (i->params != NULL) {
            saveDeleteArr(i->params);
            i->params = NULL;
        }

        // release modules loaded for this rule
        loader->releaseModule(i->module);
        
        // FIXME disable timers
    }
    
    ra->actions.clear();
    ra->lastPkt = 0;
    ra->flowKeyLen = 0;
    saveDeleteArr(ra->flowKeyList);
    ra->flist = NULL;
    ra->auto_flows = 0;
    ra->bidir = 0;
    ra->seppaths = 0;
    ra->newFlow = 0;
    ra->rule = NULL;
    if (ra->flows != NULL) {
        saveDelete(ra->flows);
    }

    return 0;
}


/* ------------------------- processPacket ------------------------- */


int PacketProcessor::processPacket(metaData_t *meta)
{

#ifdef PROFILING
    unsigned long long ini, end;
#endif
    ruleActions_t *ra;
    ppactionList_t *acts;

    AUTOLOCK(threaded, &maccess);

    // loop over all the rules matched
    for (int i = 0; i<meta->match_cnt; i++) {
        int ruleId = meta->match[i];

        ra = &rules[ruleId];

        // make sure the rule still exists
        if (!ra->actions.empty()) {
	    flowInfo_t *flow = NULL;
            //log->dlog(ch,"processing packet for Rule #%d", ruleId);

            // account number of packets and bytes for this task
            ra->packets++;
            ra->bytes += meta->cap_len;

            if (ra->auto_flows) {
	        int mval_len = 0, rmval_len = 0;
                unsigned char mvalues[1024];
		unsigned char rmvalues[1024];
		unsigned char tmp[MAX_FILTER_LEN];

                // get matching attributes from packet
                for (filterListIter_t fi = ra->flist->begin(); fi != ra->flist->end(); ++fi) {

                    unsigned char *pval = (unsigned char *) &meta->payload[ meta->offs[fi->refer] + fi->offs ]; 

                    // get masked pkt value
                    for (int i = 0; i < fi->len; i++) {
		        tmp[i] = pval[i] & fi->fdmask.getValue()[i] & fi->mask.getValue()[i];
                    }

		    // if pkt value was only one byte then shift the value 
		    // downwards according to filter def mask (e.g. 0x02 -> 1bit, 0x10 -> 4bits)
		    if (fi->len == 1 && fi->fdshift != 0) {
			tmp[0] = tmp[0] >> fi->fdshift;
		    }

		    // FIXME the following 'switch' is terribly inefficient
		    if (fi->type == "String") {
		       memcpy(&mvalues[mval_len], tmp, fi->len);
		       mval_len += fi->len;
		       // insert final terminator
		       mvalues[mval_len] = '\0';
		       mval_len++;
		    } else if (fi->type == "Binary") {
		      // insert length first
		      *((unsigned int *) &mvalues[mval_len]) = fi->len;
		      mval_len += sizeof(unsigned int);
		      memcpy(&mvalues[mval_len], tmp, fi->len);
		      mval_len += fi->len;
		    } else {
		      memcpy(&mvalues[mval_len], tmp, fi->len);
		      mval_len += fi->len;
		    }
		    
		    if (ra->bidir) {

		      if (!fi->rname.empty()) {
			// get reverse value
			unsigned char *pval = (unsigned char *) &meta->payload[ meta->offs[fi->rrefer] + fi->roffs ]; 
			// get masked pkt value
			for (int i = 0; i < fi->len; i++) {
			  tmp[i] = pval[i] & fi->fdmask.getValue()[i] & fi->mask.getValue()[i];
			}	
		      } 

		      // FIXME the following 'switch' is terribly inefficient
		      if (fi->type == "String") {
			memcpy(&rmvalues[rmval_len], tmp, fi->len);
			rmval_len += fi->len;
			// insert final terminator
			rmvalues[rmval_len] = '\0';
			rmval_len++;