0: #!/usr/bin/perl
1: =foo
2: Update: added rules
3: Update: on my darwin i set up the relevant rule with
4: 
5:     ipfw add 65500 divert 65500 tcp from me to any via en0
6: 
7: replace with your iface of choice, naturally. I haven't set this up on
8: linux (yet), but I know divert sockets were ported at some point...
9: ciao ciao!
10: 
11: This little script was hell to write, because every time i didn't do the
12: firewall rule right, and had a fix for the script, my netinfo server
13: couldn't be contacted... this ment that i couldn't run
14: 
15:     ipfw flush
16: 
17: regardless, i've finally made it. It now prioritises using a size based
18: queue, which is really nothing compared to what you can do. Suprisingly
19: it's not a resource hog, and there's still some stuff to implement, such
20: as timeout based dropping using a journal (the purger is written, but
21: not the logger), and priority queue clipping using mapl's janus heaps...
22: 
23: I guess i'll prioritize the cmp sub so that it puts tcp SYN packets, and
24: non piggybacked ACKS, aswell as short ICMP and UDP at the top
25: automatically. Then I'll prioritise by service...
26: 
27: so far only the heap has made bittorrent and perlmonks chat possible at
28: the same time, even when running several bts at a time... I must say I
29: was surprised.
30: 
31: Replace "#print" with ";print" for debugging info.
32: 
33: Unresolved bug: within the SIGIO handler recv will sometimes return
34: undefined, but the loop check doesn't pick it up, so pack complains. I
35: have no idea why.
36: 
37: Bah. Enough yak, here's the stuff:
38: =cut
39: 
40: use strict;
41: use warnings;
42: 
43: use Socket;
44: use Time::HiRes qw(time sleep alarm);
45: use Fcntl qw(F_SETOWN F_GETFL F_SETFL O_ASYNC O_NONBLOCK);
46: use Array::Heap;
47: #use Array::Heap2; # same thing, with a different namespacs
48: use NetPacket::IP qw(IP_PROTO_ICMP IP_PROTO_TCP IP_PROTO_UDP);
49: use NetPacket::TCP;
50: use NetPacket::UDP;
51: use NetPacket::ICMP;
52: use Scalar::Util qw(dualvar);
53: 
54: ### KNOWN BUGS
55: # this code is very dirty and hackish, may be resolved in the future
56: 
57: ### CHANGES
58: # added real rules to order the heaps. A simple API and examples at the bottom...
59: # currently aggressively user oriented, with a bit of space for standard unixish servers
60: 
61: ### TODO
62: # score changes dynamically based on time... Instability in heap could
63: # be negligeable at this point, as TCP is fault tolerant and UDP
64: # shouldn't care. ICMP should a have high enough priority
65: #
66: # purge using timeout, not just by clipping the queue
67: #
68: # both of these are theoretically solveable using references within the queue,
69: # so that you could undef and change the value of the referants (thingys) in
70: # the heap itself. I wonder if make_heap instead of push_heap would be much worse.
71: # This way we could take a reference to an element in the heap.
72: 
73: # aggressive purging minimizes wasted bandwidth:
74:     # have an x second maximum queue life (to be implemented)
75:     # purge heap size:
76:     #   if (overflow){
77:     #       reverse order
78:     #       reverse heapify
79:     #       reverse pop
80:     #       reverse order
81:     #   }
82:     #   
83:     #   push 
84: 
85: 
86: # user config
87: sub PORT () { 65500 }; # the port to bind on
88: sub WANTED_BYTES_PER_SECOND () { 94 * 128 }; # cap ( * 128 for kilobits, * 1024 for kilobytes )... divide by ERROR
89: sub SLEEP_INTERVAL () { 0.05 }  ## smaller = smoother, more overhead. For some perspective:
90:                                 ## 0.01 is usually the size of a time slice on unices... (linux lets you get picky)
91:                                 ## 0.03 gives good response on ~10K, 0.05 on 100K and more. the question is really
92:                                 ## how many packets do you burst before delaying for them all. If you send 100 packets
93:                                 ## a second the overhead of calling sleep will cause a significant delay
94: sub PURGE_INTERVAL () { 2 } ## how often are packets purged
95: sub PURGE_TIMEOUT () { 20 } ## how long can a packet live in the queue... only as accurate as PURGE_INTERVAL
96: sub QUEUE_LIMIT () { 32 } ## how many packets are allowed to be in the queue at any given time
97: 
98: # constants
99: sub ERROR () { 0.996 }; # clipped average on my machine, for 12 * 1024, 64 * 1024, 128 * 1024
100: sub BYTES_PER_SECOND () { WANTED_BYTES_PER_SECOND / ERROR };
101: sub SLICE () { BYTES_PER_SECOND * SLEEP_INTERVAL }; # a slice of data corresponding to SLEEP_INTERVAL
102: sub NOQUEUE () { O_ASYNC | O_NONBLOCK };
103: sub PACK_QUEUE () { "a16 a*" };
104: sub PACK_DEPT () { "d S" }; ## F: less overhead, innacurate. d: accurate, more overhead,
105:                             ## D: less overhead, accurate, needs long double support (perl -V).
106: sub PURGE_MARKERS () { int(PURGE_TIMEOUT / PURGE_INTERVAL) }
107: 
108: # variables
109: my ($dept,@dept) = (0);
110: my ($qcnt,@qjournal,@queue) = (0,((undef) x PURGE_MARKERS));
111: 
112: 
113: #print "Initializing...\n";
114: #print "SLICE=", SLICE, "\n";
115: 
116: my $rules = new Rules;
117: install_rules($rules);
118: 
119: 
120: ### hackery for priority queue implmentation
121: 
122: sub queue (@) { ## use $rules to compute score, push, janusify and pop if needed
123:     my @packets = @_;
124:     
125:     foreach my $packet (@packets){  
126:         $qcnt++;
127:         $packet = dualvar $rules->evaluate(substr($packet,16)), $packet;
128:         # add journal entry
129:     }
130:     
131:     if ($qcnt > QUEUE_LIMIT){
132:         #print "queue has exceeded limit, clipping\n";
133:         @queue = reverse @queue;
134:         make_heap_cmp { $b <=> $a } @queue;
135:         while ($qcnt > QUEUE_LIMIT){
136:             pop_heap_cmp { $b <=> $a } @queue;
137:             $qcnt--;
138:         }
139:         @queue = reverse @queue;
140:     }
141:     
142:     push_heap @queue, @packets;
143: }
144: 
145: sub dequeue (){ ### pop while defined
146:     $qcnt--;
147:     pop_heap @queue;
148: }
149: 
150: ### hackery ends
151: 
152: 
153: ## set up socket
154: socket D, PF_INET, SOCK_RAW, getprotobyname("divert") or die "$!";
155: bind D,sockaddr_in(PORT,inet_aton("0.0.0.0")) or die "$!";
156: fcntl D, F_SETOWN, $$;
157: #print "fcntl returned async ", ((fcntl D, F_GETFL, 0 & NOQUEUE) ? "on" : "off"), "\n";
158: 
159: #$SIG{ALRM} = sub {
160: #   if ($qcnt){
161: #       while (defined $qjournal[0]){
162: #           undef ${shift @qjournal}; # undefine the reference
163: #           if (--$qcnt == 1){
164: #               @queue = ();
165: #               @qjournal = ((undef) x PURGE_MARKERS);
166: #               return;
167: #           }
168: #       }; shift @qjournal; # take out one marker
169: #       push @qjournal, undef; # put another in
170: #   }
171: #};
172: #alarm PURGE_INTERVAL,PURGE_INTERVAL; # purge old packets every n
173: 
174: my ($ts,$p); # lexical for SIGIO temp usage... don't alloc/daelloc every time
175: $SIG{IO} = sub {
176:     #print time(), " SIGIO: queuing up\n";
177:     while (defined($ts = recv D, $p, 65535, 0)){    # we assume packets can come in faster than SIGIO can be called
178:                                                     # (on my machine, 128 * 1024 cap, this loops usually picks up 3-4
179:                                                     # packets), so we'll save some context switching on high load,
180:         #print "undefined p ($!)\n" unless defined $p;
181:         #print "undefined ts ($!)\n" unless defined $ts;
182:         queue(pack(PACK_QUEUE, $ts, $p));
183:     }
184: };
185: 
186: #print "Initialization complete, starting read loop\n";
187: 
188: # start loop
189: my ($to, $t, $s, $l);
190: #my ($start, $end, $total); # used to compute ERROR
191: PACKET: { if (defined ($to = recv D, $_, 65535, 0)){    # blocking read. the queue is empty. $to is reassigned
192:                                                         # because the packet could come from various rules. hack at it
193:                                                         # if it ticks you off.
194:     #print time(), " received packet\n";
195:     #print "received: " . length($to) . "\n";
196:     if ($dept < SLICE){
197:         #print time(), " dept is $dept - short circuited, should take ", length($_) / BYTES_PER_SECOND, " seconds to deliver\n";
198:         send D, $_, 0, $to;
199:         $dept += length($_);
200:         push @dept, pack(PACK_DEPT, time(), length($_) );
201:         redo PACKET;
202:     } else {
203:         #print time(), " queued (too much dept: $dept)\n";
204:         queue(pack(PACK_QUEUE, $to, $_));   # pack is about 1.5 times faster than refs (benchmark)
205:     }
206:     
207:     # the queue is not empty, or dept needs purging
208:     
209:     #print time(), " clearing up queue\n";
210:     
211:     fcntl D, F_SETFL, ((fcntl D, F_GETFL, 0)|NOQUEUE); # switch to async
212:     #print "fcntl is now noqueue ", ((fcntl D, F_GETFL, 0 & NOQUEUE) ? "on" : "off"), "\n";
213:     
214:     # use to compute ERROR
215:     #$start = time;
216:     #$total = 0;
217:     
218:     until (not $qcnt){ # until the queue is empty
219:         do {
220:             #print time(), " cleaning out and making up for dept\n";
221:             $t = time;
222:             for (my $i = 0; $i < @dept; $i++){
223:                 defined $dept[$i] or next;
224:                 ($s, $l) = unpack(PACK_DEPT, $dept[$i]);
225:                 #print time(), " diff is ", time - $s, ", ", ($l / BYTES_PER_SECOND)," diff needed queue length is $#queue, dept joural has $#dept entries ($dept)\n";
226:                 if ($t > $s + ($l / BYTES_PER_SECOND) ){
227:                     $dept -= $l;
228:                     delete($dept[$i]); # faster than splice
229:                 }
230:             }
231:             while (@dept and not exists $dept[0]){ shift @dept }; ## clean out those which are easy
232:             #print time(), " dept is now $dept\n";
233:             #print time(), " will sleep for ", $dept / BYTES_PER_SECOND,"\n" if $dept > SLICE;
234:         } while (($dept > SLICE) and sleep $dept / BYTES_PER_SECOND);   # sleep (one should suffice, but in case a sig came
235:                                                                         # (IO, ALRM are used)) until we've cleared the dept
236:         #print time(), " dept is now $dept, flushing a packet\n";
237:         
238:         my ($to,$p) = unpack(PACK_QUEUE, dequeue() );
239:         $dept += length($p);
240:         push @dept, pack(PACK_DEPT, time(), length($p) );
241:         #$total += length($p); used to compute ERROR
242:         #print time(), " sent one from queue, dept is now $dept, should take ", length($p) / BYTES_PER_SECOND, " seconds to deliver (queue left: $#queue)\n";
243:         send D, $p, 0, $to;
244:     
245:         !$qcnt ? fcntl D, F_SETFL, ((fcntl D, F_GETFL, 0)&!NOQUEUE) : redo ;    # unset async. checking here will skip checking
246:                                                                                 # until(!queue), up to the time fcntl is called.
247:                                                                                 # Then a double check is made to avoid a packet
248:                                                                                 # getting stuck in the queue while others are
249:                                                                                 # getting short circuited
250:         #print "fcntl is now noqueue ", ((fcntl D, F_GETFL, 0 & NOQUEUE) ? "on" : "off"), "\n";
251:     }
252:     
253:     # use this code to determine ERROR
254:     #$end = time;
255:     #my $bps = ($total/($end-$start));
256:     # print "during high load we sent $total bytes in ", $end-$start, " seconds, which means ", $bps, " bytes per second.\n";
257:     # print "the ratio of actual rate versus cap is ", $bps/BYTES_PER_SECOND, "\n";
258:     
259:     #print time(), " queue empty, returned to synchronious IO\n";
260:     
261:     # the queue is empty
262: } redo }
263: 
264: 
265: 1; # Keep your mother happy.
266: 
267: sub install_rules { ## the rules
268:     $_[0]->install(
269:         ### DEPENDANCIES
270:         Rule::Dependancy::Simple->new({ # basic (network unrelated) data
271:             provides => ["basic"],
272:             evaluate => sub {
273:                 my $packet = shift;
274:                 my $basic = new Dependancy::Simple;
275:                 
276:                 $basic->set("size",length($packet));
277:                 
278:                 {basic => $basic};
279:             },
280:         }),
281:         
282:         Rule::Dependancy::Simple->new({ # ip packet data
283:             provides => ["ip"],
284:             evaluate => sub { {ip => NetPacket::IP->decode($_[0]) } }
285:         }),
286:         
287:         Rule::Dependancy::Simple->new({ # tcp packet data
288:             needs => ["ip"],
289:             provides => ["tcp"],
290:             evaluate => sub {
291:                 #print "providing tcp packet dependancy\n";
292:                 ##print "got packet: ", unpack("H*",$_[0]), "\n";
293:                 ##print "Available dependancies:\n\n", do { use Data::Dumper; Dumper $_[1] },"\n";
294:                 
295:                 ($_[1]{ip}{proto} == IP_PROTO_TCP) ? {tcp => NetPacket::TCP->decode($_[1]{ip}{data}) } : {} }
296:         }),
297:         
298:         Rule::Dependancy::Simple->new({ # udp packet data
299:             needs => ["ip"],
300:             provides => ["udp"],
301:             evaluate => sub { ($_[1]{ip}{proto} == IP_PROTO_UDP) ? {udp => NetPacket::UDP->decode($_[1]{ip}{data}) } : {} }
302:         }),
303:         
304:         Rule::Dependancy::Simple->new({ # icmp packet data
305:             needs => ["ip"],
306:             provides => ["icmp"],
307:             evaluate => sub { ($_[1]{ip}{proto} == IP_PROTO_ICMP) ? {icmp => NetPacket::ICMP->decode($_[1]{ip}{data}) } : {} }
308:         }),
309:         
310:         ### RULES
311:         Rule::Simple->new({ # handle Type of Service et cetera (delay += 8, thoroughput += 5, reliability += 4, cost += 1, congestion += 2)
312:             needs => ["ip"],
313:             evaluate => sub { 0 },
314:         }),
315: 
316:         Rule::Simple->new({ # packet size
317:             needs => ["basic"],
318:             evaluate => sub {
319:                 #print "evaluating size based score adjustment\n";
320:                 length($_[1]{basic}->get("size")) ? (1.5 * log(length($_[1]{basic}->get("size")))) : 0 }
321:         }),
322:         
323:         Rule::Simple->new({ # tcp window size
324:             needs => ["tcp"],
325:             evaluate => sub {
326:                 #print "evaluating window size score adjustment\n";
327:                  $_[1]{tcp}{winsize} ? 0.1 * log($_[1]{tcp}{winsize}) : 0 }
328:         }),
329:         
330:         Rule::Simple->new({ # icmp conditional
331:             needs => ["icmp"],
332:             evaluate => sub {
333:                 #print "packet is icmp, -20\n";
334:                 -20 },
335:         }),
336:         
337:         
338:         Rule::Simple->new({ # udp conditional
339:             needs => ["udp"],
340:             evaluate => sub {
341:                 #print "packet is UDP, -6\n";
342:                  -6 },
343:         }),
344:         
345:         Rule::Simple->new({ # tcp flags
346:             needs => ["tcp"],
347:             evaluate => sub {
348:                 #print "evaluating tcp flags\n";
349:                 my $flags = $_[1]{tcp}{flags};
350:                 
351:                 my $ret = 0;
352:                 
353:                 # tcp messages with special information have varying degrees of additional importance
354:                 $ret -= 1   if $flags & FIN;
355:                 $ret -= 8   if $flags & SYN;
356:                 $ret -= 20  if $flags & RST; # attempt to help prevent waste by pushing as fast as possible. They're pretty rare anyway
357:                 $ret -= 5   if $flags & PSH;
358:                 $ret -= 2   if $flags & ACK; # packets without acks aren't as urgent
359:                 $ret -= 20  if $flags & URG;
360:             #   $ret += 0   if $flags & ECE;
361:             #   $ret += 0   if $flags & CWR;
362:                 #print "final score: $ret\n";
363:                 $ret;
364:             }
365:         }),
366:         
367:         Rule::Simple->new({ # generic (udp, tcp) port handling
368:             wants => ["tcp","udp"], # we either have tcp, or tcp
369:             evaluate => sub {
370:                 #print "evaluating port rules\n";
371:                 my $prot = (exists $_[1]->{tcp}) ? $_[1]{tcp} : $_[1]{udp};
372:                 
373:                 my $ret = 0;
374:                 
375:                 my $src = $prot->{src_port};
376:                 my $dst = $prot->{dest_port};
377:                 
378:                 #print "ports: dest=$dst, src=$src\n"; 
379:                 
380:                 SWITCH: { # source port
381:                     # unpriviliged ports
382:                     $src > 1024 and do {
383:                         $ret += 2;
384:                         
385:                         #print "source port is unpriviliged\n";
386:                         
387:                         $ret += 18, last if ($src >= 6881 and $src <= 6888);    # bittorrent
388:                         $ret += 17, last if $src == 5501;                   # hotline
389:                         $ret += 15, last if $src == 20;                     # ftp data
390:                         
391:                         last;
392:                     };
393:                     
394:                     # important services
395:                     $src == 80  and $ret -= 1, last;    # http
396:                     $src == 443 and $ret -= 1, last;    # https
397:                     $src == 143 and $ret -= 4, last;    # imap
398:                     $src == 110 and $ret -= 4, last;    # pop3
399:                     $src == 25  and $ret -= 5, last;    # smtp
400:                     $src == 22  and $ret -= 7, last;    # ssh
401:                     $src == 21  and $ret -= 6, last;    # ftp control
402:                 }
403:                 
404:                 SWITCH: { # destination port
405:                     $dst > 1024 and do {
406:                         $ret += 3;
407:                         
408:                         #print "destination port is unpriviliged\n";
409:                         
410:                         $ret += 16, last if ($dst >= 6881 and $dst <= 6888) and not ($src >= 6881 and $src <= 6888);
411:                         $ret += 15, last if $dst == 5501;
412:                         $ret += 14, last if $dst == 20;
413:                         
414:                         last;
415:                     };
416:                     
417:                     $dst == 80  and $ret -= 6, last;    # http
418:                     $dst == 443 and $ret -= 6, last;    # https
419:                     $dst == 143 and $ret -= 4, last;    # imap
420:                     $dst == 110 and $ret -= 4, last;    # pop3
421:                     $dst == 25  and $ret -= 2, last;    # smtp
422:                     $dst == 22  and $ret -= 10, last;   # ssh
423:                     $dst == 23  and $ret -= 10, last;   # telnet
424:                     $dst == 21  and $ret -= 6, last;    # ftp ctrl
425:                 }
426:                 
427:                 #print "port score: $ret\n";
428:                 
429:                 $ret;
430:             }
431:         }),
432:     )
433: }
434: 
435: package Rules; # API for joint abstraction - rules depend on common shared data, and may be added and removed.
436: 
437: # rules evaluate recursive, possibly asynchroniously in the future.
438: # once a dependancy is solved it may not be altered, and all it's children may be computed on it with no locking - methods are supposed to return static or unrelated data.
439: # a dependancy gets it's own 
440: 
441: 
442: # dependancy is either or: (more complexity may be built by creating empty dependnancy rules)
443: # needs -> a strong dependancy list. every dependancy must be met (evaluated as soon as all are met)
444: # wants -> a weak dependnancy list, at least one must be met (evaluated as soon as one is met)
445: 
446: # evaluate -> run the rule, and return either a hash of dependancy objects, or a score modification
447: 
448: # provides -> currently irrelevant. for hinting install in the future
449: 
450: sub new {
451:     bless [],shift; # dependancy tree, inverse dependancy tree, rules pending parent, execution tree
452: }
453: 
454: sub install { # clear rules that will never have all their dependancies met, and then filter for duplicates
455:     my $self = shift;
456:     
457:     # filter here
458:     
459:     #print "installing score rules\n";
460:     
461:     push @$self,@_;
462: }
463: 
464: sub evaluate { # evaluate all of the rules and return the sum
465:     my $self = shift;
466:     my $packet = shift;
467:     #no warnings; # perl doesn't like me playing with closures
468:     
469:     my %offers;
470:     my %deferred;
471:     my @ruleq;
472:     
473:     my $score = 0;
474:     
475:     #print "evaluating entire ruleset\n";
476:     
477:     foreach my $rule (@$self){
478:         my $dep = [ 0, $rule ];
479:         
480:         # build dependancy counter
481:         if ($rule->has_deps){
482:             my @needs;
483:             if ($rule->strong_deps){
484:                 @needs = grep { not exists $offers{$_} } $rule->needs;
485:                 $dep->[0] = scalar @needs;
486:             } else {
487:                 $dep->[0] = 1;
488:                 @needs = grep { $dep->[0] and (exists $offers{$_} ? (($dep->[0] = 0),undef) : 1) } $rule->needs;
489:                 $dep->[0] or @needs = ();
490:             }
491:             #print "this rule needs (@needs)\n";
492:             foreach my $dependancy (@needs){
493:                 $deferred{$dependancy} = $dep;
494:             }
495:         }
496:         
497:         push @ruleq,try($packet,\$score,\%offers,\%deferred,$dep);
498:     }
499:     
500:     my $last = scalar @ruleq;
501:     while(@ruleq){ # finish the loop
502:         #print "attempting to evaluate remaining rules\n";
503:         push @ruleq, try($packet,\$score,\%offers,\%deferred,shift @ruleq);
504:         
505:         (last == @ruleq) ? last : ($last = @ruleq); # break an infinite loop
506:     }
507:     
508:     #print "Final score is $score\n";
509:     return $score;
510:     
511:     sub try {
512:         my ($packet,$score,$offers,$deferred,$dep) = (@_);
513:         #print "trying to evaluate rule\n";
514:         if ($dep->[0] < 1){
515:             #print "all dependancies met\n";
516:             my $ret = $dep->[1]->evaluate($packet,$offers);
517:             if (ref $ret){
518:                 #print "rule introduced new offerings:";
519:                 foreach my $key (keys %{$ret}){
520:                     #print " $key,";
521:                     $offers->{$key} = $ret->{$key}; # install dependancies
522:                     
523:                     foreach my $dependant (@{$deferred->{$key}}){
524:                         $dependant->[0]--; # dependancy count goes down by one
525:                     }
526:                 }
527:                 #print "\n";
528:             } else {
529:                 #print "rule adjusted score by $ret\n";
530:                 $$score += $ret;
531:             } # don't forget this is a closure
532:             
533:             return (); # we have nothing to requeue
534:         } else {
535:             #print "unmet dependancies\n";
536:             return $dep; # requeue the current one
537:         }
538:     }
539: }
540: 
541: ## base packages for rules
542: 
543: package Rule::Simple; # a rule is something that fits in rules, and works via a certain API. a leaf in a dependancy tree
544: 
545: sub new {
546:     my $pkg = shift;
547:     bless shift, $pkg;
548: }
549: sub has_deps { (exists $_[0]{needs} or exists $_[0]{wants}) ? 1 : undef };
550: sub strong_deps { (exists $_[0]{needs}) ? 1 : undef };
551: sub needs { (exists $_[0]{needs}) ? @{$_[0]{needs}} : @{$_[0]{wants}} }
552: sub evaluate { goto &{shift->{evaluate}} }
553: 
554: package Rule::Dependancy::Simple; # a dependancy rule is something another dependancy rule or plain rule needs. a node in a dependancy tree.
555: 
556: use base "Rule::Simple"; # a simple rule that also provides();
557: sub provides { @{$_[0]{provides}} }
558: 
559: package Dependancy::Simple; # a dependancy is something a dependancy rule creates - This is just a base class for dependancy objects to work on. It contains plain values, and is basically a blessed hash
560: 
561: sub new { bless {},shift }
562: 
563: sub set { # set a value
564:     $_[0]{$_[1]} = $_[2];
565: }
566: 
567: sub get { # get a value
568:     $_[0]{$_[1]}
569: }
570: 
571: __END__