[rules-users] Implementation of my use case - what am I doing wrong?
Elran Dvir
elrand at checkpoint.com
Sun Sep 22 09:28:16 EDT 2013
Thanks for the response.
>Computationally, during the evaluation of the temporal operators.
>Why would you want to do this? (My advice to replace the temporal operators by the equivalent expressions was meant as a debugging aid, to show you where the problem with this constraint is.)
So if changing the condition to use startTimeStamp and endTimestamp caused a compilation error, How can I debug my problem? What is the equivalent of "(this meets $ce || this during $ce || this metby $ce)" without my own
startTimeStamp and endTimestamp fields?
> - Why do you use this complex declare?
> @timestamp( timestamp ) @duration( duration )
> is sufficient.
I want to protect the fields from being set outside their designated functions .I simplified it a bit in my correction.
> - Why do you use a Map and not simple fields? The DRL code is very confusing.
Because the fields of connection log may change very frequently. I don't want to be forced to compile the code with every field added.
I am getting the log as a map from outside of my application, so the translation to Log class is very easy.
The port scan is just an example. In my general use case, the user defines rules in our own language and which fields to consider, and I should translate it to DRL.
> - Why do you use two accumulate CE's when the set of matching elements is the same in both?
In the first rule, I want to collect the ids ("marker") only of the logs with unique port.
Marker is a unique identifier of a log. If I accumulate the markers in the same accumulation of port, it will collect all markers (because of their uniqueness), regardless if the belong to a log with unique port.
But thanks to your remark, I rethought my whole implementation. Thank you very much!
(I added "$portSet.contains(fieldsMap.get("port")" to markers accumulation in the first rule)
I will rewrite my fourth question with my fixes and all required data:
I am trying to identify a port scan event:
Basic event is connection log. For each combination of source_ip and destination_ip, detect a port scan event, if over 5 seconds there were more than 2 connection logs with different ports .
The event will stay open for 10 seconds and an update will be sent for any new port detected. Every update will contain the count of connection logs combining it and their id ("marker").
This is my drl file:
----------------------------------------------------------------------------------------------------
package test;
import correlation.impl.drools.Log
import correlation.impl.drools.CorrelatedEvent
global correlation.server.EventsHandler externalEventsHandler;
declare Log
@role( event)
end
declare CorrelatedEvent
//@propertyReactive
@role( event)
@timestamp( getTimestamp())
@expires( 10s )
@duration( getDuration() )
end
// this rule will create a "Port Scan" event if none exist for this group-by values
rule "Create Port Scan Event"
dialect "java" //dialect can be either mvel or java for the RHS (only java dialect support generic containers)
no-loop //this means - do not revaluate rules if the events were altered in memory
when
$log : Log()
accumulate( Log( this after[0s,5s] $log, fieldsMap.get("src") == $log.fieldsMap.get("src") , fieldsMap.get("dst") == $log.fieldsMap.get("dst"), $port : fieldsMap.get("port"));
$portSet : collectSet($port);
$portSet.size > 2 )
accumulate( Log( this after[0s,5s] $log, fieldsMap.get("src") == $log.fieldsMap.get("src") , fieldsMap.get("dst") == $log.fieldsMap.get("dst") , $portSet.contains(fieldsMap.get("port")) , $marker : fieldsMap.get("marker"));
$markerSet : collectSet($marker))
not CorrelatedEvent(getName() == "portScan" , fieldsMap.get("src") == $log.fieldsMap.get("src") , fieldsMap.get("dst") == $log.fieldsMap.get("dst"))
then
CorrelatedEvent $ce = new CorrelatedEvent();
$ce.setName("portScan");
$ce.setEventsHandler(externalEventsHandler);
$ce.setDurationInSec(10);
$ce.fieldsMap.put("src", $log.fieldsMap.get("src"));
$ce.fieldsMap.put("dst", $log.fieldsMap.get("dst"));
$ce.endUpdate($markerSet, $portSet);
insert($ce);
end
rule "Create Port Scan Event - update"
dialect "java" //dialect can be either mvel or java for the RHS (only java dialect support generic containers)
no-loop
when
$ce: CorrelatedEvent(getName() == "portScan", $portSet: getUniqueSet())
$log: Log(fieldsMap.get("src") == $ce.fieldsMap.get("src") , fieldsMap.get("dst") == $ce.fieldsMap.get("dst") , !$portSet.contains(fieldsMap.get("port")), this after $ce.getStartTime() , this before $ce.getEndTime())
then
modify( $ce ) {endUpdate($log.fieldsMap.get("marker").toString(), $log.fieldsMap.get("port"))}
end
------------------------------------------------------------------------------------------------------------------------------------
I test it like this:
I insert a connection log and fire the rules every second. I have 25 logs with the same "src" and "dst", but each has different (serial) "port" and "marker".
So after 12-13 logs, I expect to identify a new event with another consecutive 3 logs.
In each rule's RHS, I print the rule fired and the ports accumulated so far.
With existing implementation, I see the following output at 14th second:
rule fired: Create Port Scan Event - update
portSet: [1, 7, 11, 8, 5, 6, 3, 10, 4, 9, 12, 2]
rule fired: Create Port Scan Event
portSet: [13, 11, 12]
As we can see, the first rule processes logs already processed by the second rule (11,12).
After examining the first rule, I understood this behavior.
I decided to change the order of conditions in the LHS of the first rule by moving "not CorrelatedEvent..." to be the second condition.
But then I get the following output after the first 4 logs:
rule fired: Create Port Scan Event
portSet: []
rule fired: Create Port Scan Event - update
portSet: [1, 3, 4, 2]
Why is that? Where the first 3 events "disappeared"? How $portSet is empty with the condition $portSet.size > 2?
Thanks a lot!
---------------------------------------------------------------
Log class:
----------------------------------------------------------------
public class Log {
public HashMap<String, Object> fieldsMap = new HashMap<>();
}
-------------------------------------------------------------------------------------------
CorrelatedEvent class:
---------------------------------------------------------------------------------------------
public class CorrelatedEvent
{
public Map<String, Object> fieldsMap;
public int duration; //in ms
public long timestamp;
private String name;
private Set<String> markersSet;
private Set uniqueSet;
private long logsCount;
private Calendar startTime;
private Calendar endTime;
private EventsHandler eventsHandler;
public CorrelatedEvent()
{
startTime = Calendar.getInstance();
timestamp = startTime.getTime().getTime();
endTime = Calendar.getInstance();
endTime.setTime(startTime.getTime());
duration = 0;
fieldsMap = new HashMap<>();
markersSet = Collections.newSetFromMap(new ConcurrentHashMap<String, Boolean>());
uniqueSet = Collections.newSetFromMap(new ConcurrentHashMap<Object, Boolean>());
logsCount = 0;
}
public Set getUniqueSet()
{
return uniqueSet;
}
public void endUpdate(Set<String> markersSet, Set uniqueSet)
{
this.markersSet.addAll(markersSet);
this.uniqueSet.addAll(uniqueSet);
handleUpdate();
}
public void endUpdate(String marker, Object uniqueValue)
{
markersSet.add(marker);
uniqueSet.add(uniqueValue);
handleUpdate();
}
private void handleUpdate()
{
if (markersSet.size() != uniqueSet.size())
return;
if (markersSet.size() > logsCount) {
logsCount = this.markersSet.size();
if (eventsHandler == null)
return;
Map<String, Object> clonedFieldsMap = getClonedFieldsMap();
clonedFieldsMap.put("markers", markersSet.toString());
clonedFieldsMap.put("count", logsCount);
eventsHandler.handleEvent(clonedFieldsMap);
}
}
private Map<String, Object> getClonedFieldsMap()
{
Map<String, Object> clonedFieldsMap = new HashMap<>();
clonedFieldsMap.putAll(fieldsMap);
return clonedFieldsMap;
}
public Date getStartTime()
{
return startTime.getTime();
}
public void setDurationInSec(int duration)
{
this.duration = duration * 1000;
endTime.setTime(startTime.getTime());
endTime.add(Calendar.SECOND, duration);
}
public Date getEndTime()
{
return endTime.getTime();
}
public long getTimestamp()
{
return timestamp;
}
public int getDuration()
{
return duration;
}
public String getName()
{
return name;
}
public void setName(String name)
{
this.name = name;
fieldsMap.put("name", name);
}
public EventsHandler getEventsHandler()
{
return eventsHandler;
}
public void setEventsHandler(EventsHandler eventsHandler)
{
this.eventsHandler = eventsHandler;
}
}
------------------------------------------------------------------------------------------------
-----Original Message-----
From: rules-users-bounces at lists.jboss.org [mailto:rules-users-bounces at lists.jboss.org] On Behalf Of Wolfgang Laun
Sent: Thursday, September 19, 2013 10:00 AM
To: Rules Users List
Subject: Re: [rules-users] Implementation of my use case - what am I doing wrong?
On 17/09/2013, Elran Dvir <elrand at checkpoint.com> wrote:
> Thanks again.
>
> I don't have startTimestamp and endTimestamp fields. I assume these
> fields are created on runtime bases on duration and timestamp
> attributes. Isn't it?
Computationally, during the evaluation of the temporal operators.
> If I need to define them myself, what is the advantage of defining
> timestamp and duration attributes?
Why would you want to do this? (My advice to replace the temporal operators by the equivalent expressions was meant as a debugging aid, to show you where the problem with this constraint is.)
>
> I'll try to organize the fourth question:
I can't answer your question - it's impossible to tell what might happen without the code for CorrelatedEvent.java. For reproducing this, also the code for Log.java should be available. A couple of
remarks:
- Why do you use this complex declare?
@timestamp( timestamp ) @duration( duration )
is sufficient.
- Why do you use a Map and not simple fields? The DRL code is very confusing.
- Why do you use two accumulate CE's when the set of matching elements is the same in both?
-W
> I am trying to identify a port scan event:
> Basic event is connection log. For each combination of source_ip and
> destination_ip, detect a port scan event, if over 5 seconds there were
> more than 2 connection logs with different ports .
> The event will stay open for 10 seconds and an update will be sent for
> any new port detected. Every update will contain the count of
> connection logs combining it and their id ("marker").
>
> This is my drl file:
> ----------------------------------------------------------------------
> ------------------------------
> package test;
>
> import correlation.impl.drools.Log
> import correlation.impl.drools.CorrelatedEvent
>
> global correlation.server.EventsHandler externalEventsHandler;
>
> declare Log
> @role( event)
> end
>
> declare CorrelatedEvent
> @role( event)
> @timestamp( getTimestamp().getTime() )
> @expires( 10s )
> @duration( getDuration() )
> end
>
> // this rule will create a "Port Scan" event if none exist for this
> group-by values rule "Create Port Scan Event"
> dialect "java"
> no-loop
> when
> $log : Log() from entry-point "Log stream" //get all the logs in the
> last
> 5 seconds
> accumulate( Log( this after[0s,5s] $log, fieldsMap.get("src") ==
> $log.fieldsMap.get("src") , fieldsMap.get("dst") ==
> $log.fieldsMap.get("dst"), $port : fieldsMap.get("port")) from
> entry-point "Log stream";
> $portSet : collectSet($port);
> $portSet.size > 2 )
> accumulate( Log( this after[0s,5s] $log, fieldsMap.get("src") ==
> $log.fieldsMap.get("src") , fieldsMap.get("dst") ==
> $log.fieldsMap.get("dst"), $marker : fieldsMap.get("marker")) from
> entry-point "Log stream";
> $markerSet : collectSet($marker))
> not CorrelatedEvent(getName() == "portScan" , fieldsMap.get("src")
> ==
> $log.fieldsMap.get("src") , fieldsMap.get("dst") ==
> $log.fieldsMap.get("dst"))
> then
> System.out.println(drools.getRule().getName());
>
> CorrelatedEvent $ce = new CorrelatedEvent();
> $ce.setName("portScan");
> $ce.setEventsHandler(externalEventsHandler);
> $ce.setDurationInSec(10);
> $ce.fieldsMap.put("src", $log.fieldsMap.get("src"));
> $ce.fieldsMap.put("dst", $log.fieldsMap.get("dst"));
> $ce.endUpdate($markerSet);
>
> insert($ce);
> end
>
> rule "Create Port Scan Event - update"
> dialect "java"
> no-loop
> when
> $ce: CorrelatedEvent(getName() == "portScan")
> accumulate( Log(fieldsMap.get("src") == $ce.fieldsMap.get("src") ,
> fieldsMap.get("dst") == $ce.fieldsMap.get("dst") , $port :
> fieldsMap.get("port") , this after $ce.getStartTime() , this before
> $ce.getEndTime()) from entry-point "Log stream";
> $portSet : collectSet($port);
> $portSet.size > 0 )
> accumulate( Log(fieldsMap.get("src") == $ce.fieldsMap.get("src") ,
> fieldsMap.get("dst") == $ce.fieldsMap.get("dst") , $marker :
> fieldsMap.get("marker") , this after $ce.getStartTime() , this before
> $ce.getEndTime()) from entry-point "Log stream";
> $markerSet : collectSet($marker)) then
> System.out.println(drools.getRule().getName());
>
> modify( $ce ) {endUpdate($markerSet)} end
> ----------------------------------------------------------------------
> --------------------------------------------------------------
> I test it like this:
> I insert a connection log and fire the rules every second. I have 25
> logs with the same "src" and "dst", but each has different (serial)
> "port" and "marker".
> So after 12-13 logs, I expect to identify a new event with another
> consecutive 3 logs.
> In each rule's RHS, I print the rule fired and the port set of logs
> triggering it.
> With existing implementation, I see the following output at 14th second:
>
> rule fired: Create Port Scan Event - update
> portSet: [10, 7, 6, 5, 4, 9, 8, 11, 12]
>
> rule fired: Create Port Scan Event
> portSet: [13, 11, 12]
>
> As we can see, the first rule processes logs already processed by the
> second rule.
> After examining the first rule, I understood this behavior.
> I decided to change the order of conditions in the LHS of the first
> rule by moving "not CorrelatedEvent..." to be the second condition.
> But then I get the following output after the first 4 logs:
>
> rule fired: Create Port Scan Event
> portSet: []
>
> rule fired: Create Port Scan Event - update
> portSet: [4]
>
> Why is that? Where the first 3 events "disappeared"? How $portSet is
> empty with the condition $portSet.size > 2?
>
> Thanks a lot.
>
> -----Original Message-----
> From: rules-users-bounces at lists.jboss.org
> [mailto:rules-users-bounces at lists.jboss.org] On Behalf Of Wolfgang
> Laun
> Sent: Tuesday, September 17, 2013 2:08 PM
> To: Rules Users List
> Subject: Re: [rules-users] Implementation of my use case - what am I
> doing wrong?
>
> On 17/09/2013, Elran Dvir <elrand at checkpoint.com> wrote:
>> Thanks for the quick response.
>>
>> I have some more questions:
>>
>> 1. As I understand it, the timestamp attribute should be long type
>> representing the milliseconds since January 1, 1970, 00:00:00 GMT. Am
>> I right?
>
> Not necessarily. The interpretation of this long value is up to you -
> it could mean days since the foundation of Rome (753 BC).
>
>> 2. As I understand it, the duration attribute should be in
>> milliseconds. I fixed it accordingly. Am I right?
>
> Use the same unit as the timestamp.
>
>> 3. When I replaced "(this meets $ce || this during $ce || this metby
>> $ce)"
>> with "$ce.startTimestamp <= startTimestamp , endTimestamp <=
>> $ce.endTimestamp"
>> I got the following drools compile exceptions:
>>
>> Unable to Analyse Expression $ce.startTimestamp:
>> [Error: unable to resolve method using strict-mode:
>> com.checkpoint.correlation.impl.drools.CorrelatedEvent.startTimestamp()]
>> [Near : {... $ce.startTimestamp ....}]
>> ^
>> [Line: 61, Column: 28] : [Rule name='Create Port Scan Event -
>> update']
>>
>> Unable to Analyse Expression $ce.startTimestamp <= startTimestamp:
>> [Error: unable to resolve method using strict-mode:
>> com.checkpoint.correlation.impl.drools.CorrelatedEvent.startTimestamp()]
>> [Near : {... $ce.startTimestamp <= startTimesta ....}]
>> ^
>> [Line: 61, Column: 28] : [Rule name='Create Port Scan Event -
>> update']
>>
>> Unable to Analyse Expression endTimestamp <= $ce.endTimestamp:
>> [Error: unable to resolve method using strict-mode:
>> com.checkpoint.correlation.impl.drools.CpLog.endTimestamp()]
>> [Near : {... endTimestamp <= $ce.endTimesta ....}]
>> ^
>> [Line: 61, Column: 28] : [Rule name='Create Port Scan Event -
>> update']
>>
>> Unable to Analyse Expression $ce.startTimestamp:
>> [Error: unable to resolve method using strict-mode:
>> com.checkpoint.correlation.impl.drools.CorrelatedEvent.startTimestamp()]
>> [Near : {... $ce.startTimestamp ....}]
>>
>> Why?
>
> Do you have fields startTimestamp and endTimestamp?
>
>> 4. I tested my working implementation of temporal relation in rule
>> "Create Port Scan Event - update" ("this after $ce.getStartTime() ,
>> this before
>> $ce.getEndTime()") .
> [snip]
>
>>
>> Why is that? Where the first 3 events disappeared? How "portSet"
>> is empty with the condition $portSet.size > 2?
>
> Sorry, you've lost me here. I can't see what's going on from this
> unorganized set of snippets - and please don't suppose that people
> keep old mails or have the time to dig in the archives.
>
> -W
>
>>
>> Thanks a lot.
>>
>> -----Original Message-----
>> From: rules-users-bounces at lists.jboss.org
>> [mailto:rules-users-bounces at lists.jboss.org] On Behalf Of Wolfgang
>> Laun
>> Sent: Sunday, September 15, 2013 8:08 PM
>> To: Rules Users List
>> Subject: Re: [rules-users] Implementation of my use case - what am I
>> doing wrong?
>>
>> On 15/09/2013, Elran Dvir <elrand at checkpoint.com> wrote:
>>
>>> my questions:
>>>
>>> 1) If I have only one stream of data , can I omit the use of entry
>>> point and insert logs to the session ? Or the use of entry points is
>>> mandatory in Drools Fusion?
>>
>> Yes. No. An entry point is just an additional attribute added "on the
>> fly", where you don't have a source identification in the pojo.
>>
>>>
>>> 2) When I tested it with matching data, rule "Create Port Scan
>>> Event
>>> -
>>> update" was never fired. When I replaced "(this meets $ce || this
>>> during $ce
>>> || this metby $ce)" with "this after $ce.getStartTime() , this
>>> || before
>>> $ce.getEndTime()" everything worked fine.
>>> Why?
>>
>> Just take the constraints and replace the temporal operator by its
>> definition in the "Fusion" manual and use a little elementary math:
>>
>> A meets || A during B || A metby B becomes
>> abs( B.startTimestamp - A.endTimestamp ) == 0 ||
>> B.startTimestamp < A.startTimestamp && A.endTimestamp <
>> B.endTimestamp || abs( A.startTimestamp - B.endTimestamp ) == 0 becomes
>> ...
>>
>>
>>>
>>> 3) I tried to use sliding windows in rule "Create Port Scan Event"
>>> and
>>> an exception was thrown at runtime. I decided to use "this
>>> after[0s,5s] $log" instead. Is it correct?
>>
>> A sliding window is not the same as the temporal relation of two
>> events. If the rule does what it ought to, I'd say, yes, it is correct.
>>
>>>
>>> 4) Is my basic Implementation correct?
>>
>> A bit much to ask, don't you think?
>>
>> -W
>> _______________________________________________
>> rules-users mailing list
>> rules-users at lists.jboss.org
>> https://lists.jboss.org/mailman/listinfo/rules-users
>>
>> Email secured by Check Point
>>
>> _______________________________________________
>> rules-users mailing list
>> rules-users at lists.jboss.org
>> https://lists.jboss.org/mailman/listinfo/rules-users
>>
> _______________________________________________
> rules-users mailing list
> rules-users at lists.jboss.org
> https://lists.jboss.org/mailman/listinfo/rules-users
>
> Email secured by Check Point
>
> _______________________________________________
> rules-users mailing list
> rules-users at lists.jboss.org
> https://lists.jboss.org/mailman/listinfo/rules-users
>
_______________________________________________
rules-users mailing list
rules-users at lists.jboss.org
https://lists.jboss.org/mailman/listinfo/rules-users
Email secured by Check Point
More information about the rules-users
mailing list