ESA rule - left outer join and aggregation

[JohnDoe1]

Hello all,

I have a advanced EPL rule to detect horizontal network scan which works fine.

module scan;
create constant variable string[] whitelist  = {
    '1.2.3.4',     /* exception 1*/
    '1.2.3.5',     /* exception 2 */
};
@Name('scan')
@RSAAlert(oneInSeconds=0, identifiers={"ip_src"})
SELECT * FROM Event (
        medium = 32
    AND 
        ip_src IS NOT NULL
    AND
        ip_dst IS NOT NULL
    AND
        ip_dstport IS NOT NULL
    AND
        ip_src NOT IN (whitelist)
).std:groupwin(ip_src, ip_dstport)
.std:unique(ip_dst)
.win:time_batch(60 seconds)
GROUP BY 
    ip_src, ip_dstport 
HAVING 
    count(ip_dst) >= 100;

Now, I would like to improve  the whitelisting to do the exclusion withlisting not only based on ip_src, but based on ip_src and the corresponding ip_dstport.

Ideally, I want to keep a simple statement of the whitelist like this:

create constant variable Object[] whitelist  = {
    {'1.2.3.4',80},     /* exception HTTP 1*/
    {'1.2.3.5',22}      /* exception SSH 2 */
};

What is the best way to achieve this?

Should I create a schema or a name windows or a hashmap and do a left outer join? (if yes, how to deal with aggregate and left outer join?)

[MaximilianoCitt]

Hey John, do you have resolved this?

May be can you share your experience?

[robertofasciani]

Hello John,

can you try use the In-Memory Table as Enrichment Source (Alerting: Configure In-Memory Table as Enrichment Source )?

If you configure an Adhoc In-Memory Table with the User-Defined Table Name = my_scan_wl

and use a CSV like this:

ip_src string, ip_dstport integer

1.2.3.4,80

1.2.3.5,22

Then you will use this rule:

@Name('scan')
@UsesEnrichment(name = 'my_scan_wl') //10.6.1.1 and later
@RSAAlert
SELECT * FROM Event (
   medium = 32
   AND ip_src IS NOT NULL
   AND ip_dst IS NOT NULL
   AND ip_dstport IS NOT NULL
   AND NOT EXISTS
           (SELECT * FROM my_scan_wl
                              WHERE ip_src = Event.ip_src  AND ip_dstport = Event.ip_dstport
             )
).std:groupwin(ip_src, ip_dstport)
.std:unique(ip_dst)
.win:time_batch(60 seconds)
GROUP BY ip_src, ip_dstport
HAVING count(ip_dst) >= 100;

[UtsavSejpal]

Hey Mate,

Since I wanted to whitelist specific pair of source IP - destination IP, I have made csv file something like below:

ip_src string, ip_dst string

1.2.3.4

1.2.3.5

Hope that's right. 

Best Regards,

Utsav Sejpal