PVE/HTTPServer.pm

   1 package PVE::HTTPServer;
   2
   3 use strict;
   4 use warnings;
   5 use Time::HiRes qw(usleep ualarm gettimeofday tv_interval);
   6 use Socket qw(IPPROTO_TCP TCP_NODELAY SOMAXCONN);
   7 use POSIX qw(strftime EINTR EAGAIN);
   8 use Fcntl;
   9 use IO::File;
  10 use File::stat qw();
  11 use Digest::MD5;
  12 # use AnyEvent::Strict; # only use this for debugging
  13 use AnyEvent::Util qw(guard fh_nonblocking WSAEWOULDBLOCK WSAEINPROGRESS);
  14 use AnyEvent::Socket;
  15 use AnyEvent::Handle;
  16 use Net::SSLeay;
  17 use AnyEvent::TLS;
  18 use AnyEvent::IO;
  19 use AnyEvent::HTTP;
  20 use Fcntl ();
  21 use Compress::Zlib;
  22 use PVE::SafeSyslog;
  23 use PVE::INotify;
  24 use PVE::RPCEnvironment;
  25 use PVE::REST;
  26
  27 use Net::IP;
  28 use URI;
  29 use HTTP::Status qw(:constants);
  30 use HTTP::Headers;
  31 use HTTP::Response;
  32 use Data::Dumper;
  33
  34 my $limit_max_headers = 30;
  35 my $limit_max_header_size = 8*1024;
  36 my $limit_max_post = 16*1024;
  37
  38
  39 my $known_methods = {
  40     GET => 1,
  41     POST => 1,
  42     PUT => 1,
  43     DELETE => 1,
  44 };
  45
  46 my $baseuri = "/api2";
  47
  48 sub split_abs_uri {
  49     my ($abs_uri) = @_;
  50
  51     my ($format, $rel_uri) = $abs_uri =~ m/^\Q$baseuri\E\/+(html|text|json|extjs|png|htmljs|spiceconfig)(\/.*)?$/;
  52     $rel_uri = '/' if !$rel_uri;
  53
  54     return wantarray ? ($rel_uri, $format) : $rel_uri;
  55 }
  56
  57 sub log_request {
  58     my ($self, $reqstate) = @_;
  59
  60     my $loginfo = $reqstate->{log};
  61
  62     # like apache2 common log format
  63     # LogFormat "%h %l %u %t \"%r\" %>s %b \"%{Referer}i\" \"%{User-agent}i\""
  64
  65     return if $loginfo->{written}; # avoid duplicate logs
  66     $loginfo->{written} = 1;
  67
  68     my $peerip = $reqstate->{peer_host} || '-';
  69     my $userid = $loginfo->{userid} || '-';
  70     my $content_length = defined($loginfo->{content_length}) ? $loginfo->{content_length} : '-';
  71     my $code =  $loginfo->{code} || 500;
  72     my $requestline = $loginfo->{requestline} || '-';
  73     my $timestr = strftime("%d/%b/%Y:%H:%M:%S %z", localtime());
  74
  75     my $msg = "$peerip - $userid [$timestr] \"$requestline\" $code $content_length\n";
  76
  77     $self->write_log($msg);
  78 }
  79
  80 sub log_aborted_request {
  81     my ($self, $reqstate, $error) = @_;
  82
  83     my $r = $reqstate->{request};
  84     return if !$r; # no active request
  85
  86     if ($error) {
  87         syslog("err", "problem with client $reqstate->{peer_host}; $error");
  88     }
  89
  90     $self->log_request($reqstate);
  91 }
  92
  93 sub cleanup_reqstate {
  94     my ($reqstate) = @_;
  95
  96     delete $reqstate->{log};
  97     delete $reqstate->{request};
  98     delete $reqstate->{proto};
  99     delete $reqstate->{accept_gzip};
 100
 101     if ($reqstate->{tmpfilename}) {
 102         unlink $reqstate->{tmpfilename};
 103         delete $reqstate->{tmpfilename};
 104     }
 105 }
 106
 107 sub client_do_disconnect {
 108     my ($self, $reqstate) = @_;
 109
 110     cleanup_reqstate($reqstate);
 111
 112     my $shutdown_hdl = sub {
 113         my $hdl = shift;
 114
 115         shutdown($hdl->{fh}, 1);
 116         # clear all handlers
 117         $hdl->on_drain(undef);
 118         $hdl->on_read(undef);
 119         $hdl->on_eof(undef);
 120     };
 121
 122     if (my $proxyhdl = delete $reqstate->{proxyhdl}) {
 123         &$shutdown_hdl($proxyhdl);
 124     }
 125
 126     my $hdl = delete $reqstate->{hdl};
 127
 128     if (!$hdl) {
 129         syslog('err', "detected empty handle");
 130         return;
 131     }
 132
 133     print "close connection $hdl\n" if $self->{debug};
 134
 135     &$shutdown_hdl($hdl);
 136
 137     $self->{conn_count}--;
 138
 139     print "$$: CLOSE FH" .  $hdl->{fh}->fileno() . " CONN$self->{conn_count}\n" if $self->{debug};
 140 }
 141
 142 sub finish_response {
 143     my ($self, $reqstate) = @_;
 144
 145     my $hdl = $reqstate->{hdl};
 146
 147     cleanup_reqstate($reqstate);
 148
 149     if (!$self->{end_loop} && $reqstate->{keep_alive} > 0) {
 150         # print "KEEPALIVE $reqstate->{keep_alive}\n" if $self->{debug};
 151         $hdl->on_read(sub {
 152             eval { $self->push_request_header($reqstate); };
 153             warn $@ if $@;
 154         });
 155     } else {
 156         $hdl->on_drain (sub {
 157             eval {
 158                 $self->client_do_disconnect($reqstate);
 159             };
 160             warn $@ if $@;
 161         });
 162     }
 163 }
 164
 165 sub response {
 166     my ($self, $reqstate, $resp, $mtime, $nocomp) = @_;
 167
 168     #print "$$: send response: " . Dumper($resp);
 169
 170     # activate timeout
 171     $reqstate->{hdl}->timeout_reset();
 172     $reqstate->{hdl}->timeout($self->{timeout});
 173
 174     $nocomp = 1 if !$reqstate->{accept_gzip};
 175
 176     my $code = $resp->code;
 177     my $msg = $resp->message || HTTP::Status::status_message($code);
 178     ($msg) = $msg =~m/^(.*)$/m;
 179     my $content = $resp->content;
 180
 181     if ($code =~ /^(1\d\d|[23]04)$/) {
 182         # make sure content we have no content
 183         $content = "";
 184     }
 185
 186     $reqstate->{keep_alive} = 0 if ($code >= 400) || $self->{end_loop};
 187
 188     $reqstate->{log}->{code} = $code;
 189
 190     my $proto = $reqstate->{proto} ? $reqstate->{proto}->{str} : 'HTTP/1.0';
 191     my $res = "$proto $code $msg\015\012";
 192
 193     my $ctime = time();
 194     my $date = HTTP::Date::time2str($ctime);
 195     $resp->header('Date' => $date);
 196     if ($mtime) {
 197         $resp->header('Last-Modified' => HTTP::Date::time2str($mtime));
 198     } else {
 199         $resp->header('Expires' => $date);
 200         $resp->header('Cache-Control' => "max-age=0");
 201         $resp->header("Pragma", "no-cache");
 202     }
 203
 204     $resp->header('Server' => "pve-api-daemon/3.0");
 205
 206     my $content_length;
 207     if ($content) {
 208
 209         $content_length = length($content);
 210
 211         if (!$nocomp && ($content_length > 1024)) {
 212             my $comp = Compress::Zlib::memGzip($content);
 213             $resp->header('Content-Encoding', 'gzip');
 214             $content = $comp;
 215             $content_length = length($content);
 216         }
 217         $resp->header("Content-Length" => $content_length);
 218         $reqstate->{log}->{content_length} = $content_length;
 219
 220     } else {
 221         $resp->remove_header("Content-Length");
 222     }
 223
 224     if ($reqstate->{keep_alive} > 0) {
 225         $resp->push_header('Connection' => 'Keep-Alive');
 226     } else {
 227         $resp->header('Connection' => 'close');
 228     }
 229
 230     $res .= $resp->headers_as_string("\015\012");
 231     #print "SEND(without content) $res\n" if $self->{debug};
 232
 233     $res .= "\015\012";
 234     $res .= $content if $content;
 235
 236     $self->log_request($reqstate, $reqstate->{request});
 237
 238     $reqstate->{hdl}->push_write($res);
 239     $self->finish_response($reqstate);
 240 }
 241
 242 sub error {
 243     my ($self, $reqstate, $code, $msg, $hdr, $content) = @_;
 244
 245     eval {
 246         my $resp = HTTP::Response->new($code, $msg, $hdr, $content);
 247         $self->response($reqstate, $resp);
 248     };
 249     warn $@ if $@;
 250 }
 251
 252 sub send_file_start {
 253     my ($self, $reqstate, $filename) = @_;
 254
 255     eval {
 256         # print "SEND FILE $filename\n";
 257         # Note: aio_load() this is not really async unless we use IO::AIO!
 258         eval {
 259
 260             my $r = $reqstate->{request};
 261
 262             my $fh = IO::File->new($filename, '<') ||
 263                 die "$!\n";
 264             my $stat = File::stat::stat($fh) ||
 265                 die "$!\n";
 266
 267             my $mtime = $stat->mtime;
 268
 269             if (my $ifmod = $r->header('if-modified-since')) {
 270                 my $iftime = HTTP::Date::str2time($ifmod);
 271                 if ($mtime <= $iftime) {
 272                     my $resp = HTTP::Response->new(304, "NOT MODIFIED");
 273                     $self->response($reqstate, $resp, $mtime);
 274                     return;
 275                 }
 276             }
 277
 278             my $data;
 279             my $len = sysread($fh, $data,  $stat->size);
 280             die "got short file\n" if !defined($len) || $len != $stat->size;
 281
 282             my $ct;
 283             my $nocomp;
 284             if ($filename =~ m/\.css$/) {
 285                 $ct = 'text/css';
 286             } elsif ($filename =~ m/\.js$/) {
 287                 $ct = 'application/javascript';
 288             } elsif ($filename =~ m/\.png$/) {
 289                 $ct = 'image/png';
 290                 $nocomp = 1;
 291             } elsif ($filename =~ m/\.ico$/) {
 292                 $ct = 'image/x-icon';
 293                 $nocomp = 1;
 294             } elsif ($filename =~ m/\.gif$/) {
 295                 $ct = 'image/gif';
 296                 $nocomp = 1;
 297             } elsif ($filename =~ m/\.jar$/) {
 298                 $ct = 'application/java-archive';
 299                 $nocomp = 1;
 300             } else {
 301                 die "unable to detect content type";
 302             }
 303
 304             my $header = HTTP::Headers->new(Content_Type => $ct);
 305             my $resp = HTTP::Response->new(200, "OK", $header, $data);
 306             $self->response($reqstate, $resp, $mtime, $nocomp);
 307         };
 308         if (my $err = $@) {
 309             $self->error($reqstate, 501, $err);
 310         }
 311     };
 312
 313     warn $@ if $@;
 314 }
 315
 316 sub proxy_request {
 317     my ($self, $reqstate, $clientip, $host, $method, $uri, $ticket, $token, $params) = @_;
 318
 319     eval {
 320         my $target;
 321         my $keep_alive = 1;
 322         if ($host eq 'localhost') {
 323             $target = "http://$host:85$uri";
 324             # keep alive for localhost is not worth (connection setup is about 0.2ms)
 325             $keep_alive = 0;
 326         } else {
 327             $target = "https://$host:8006$uri";
 328         }
 329
 330         my $headers = {
 331             PVEDisableProxy => 'true',
 332             PVEClientIP => $clientip,
 333         };
 334
 335         my $cookie_name = 'PVEAuthCookie';
 336
 337         $headers->{'cookie'} = PVE::REST::create_auth_cookie($ticket) if $ticket;
 338         $headers->{'CSRFPreventionToken'} = $token if $token;
 339         $headers->{'Accept-Encoding'} = 'gzip' if $reqstate->{accept_gzip};
 340
 341         my $content;
 342
 343         if  ($method eq 'POST' || $method eq 'PUT') {
 344             $headers->{'Content-Type'} = 'application/x-www-form-urlencoded';
 345             # use URI object to format application/x-www-form-urlencoded content.
 346             my $url = URI->new('http:');
 347             $url->query_form(%$params);
 348             $content = $url->query;
 349             if (defined($content)) {
 350                 $headers->{'Content-Length'} = length($content);
 351             }
 352         }
 353
 354         my $w; $w = http_request(
 355             $method => $target,
 356             headers => $headers,
 357             timeout => 30,
 358             recurse => 0,
 359             proxy => undef, # avoid use of $ENV{HTTP_PROXY}
 360             keepalive => $keep_alive,
 361             body => $content,
 362             tls_ctx => $self->{tls_ctx},
 363             sub {
 364                 my ($body, $hdr) = @_;
 365
 366                 undef $w;
 367
 368                 if (!$reqstate->{hdl}) {
 369                     warn "proxy detected vanished client connection\n";
 370                     return;
 371                 }
 372
 373                 eval {
 374                     my $code = delete $hdr->{Status};
 375                     my $msg = delete $hdr->{Reason};
 376                     delete $hdr->{URL};
 377                     delete $hdr->{HTTPVersion};
 378                     my $header = HTTP::Headers->new(%$hdr);
 379                     my $resp = HTTP::Response->new($code, $msg, $header, $body);
 380                     # Note: disable compression, because body is already compressed
 381                     $self->response($reqstate, $resp, undef, 1);
 382                 };
 383                 warn $@ if $@;
 384             });
 385     };
 386     warn $@ if $@;
 387 }
 388
 389 # return arrays as \0 separated strings (like CGI.pm)
 390 sub decode_urlencoded {
 391     my ($data) = @_;
 392
 393     my $res = {};
 394
 395     return $res if !$data;
 396
 397     foreach my $kv (split(/[\&\;]/, $data)) {
 398         my ($k, $v) = split(/=/, $kv);
 399         $k =~s/\+/ /g;
 400         $k =~ s/%([0-9a-fA-F][0-9a-fA-F])/chr(hex($1))/eg;
 401         $v =~s/\+/ /g;
 402         $v =~ s/%([0-9a-fA-F][0-9a-fA-F])/chr(hex($1))/eg;
 403
 404         if (defined(my $old = $res->{$k})) {
 405             $res->{$k} = "$old\0$v";
 406         } else {
 407             $res->{$k} = $v;
 408         }
 409     }
 410     return $res;
 411 }
 412
 413 sub extract_params {
 414     my ($r, $method) = @_;
 415
 416     my $params = {};
 417
 418     if ($method eq 'PUT' || $method eq 'POST') {
 419         $params = decode_urlencoded($r->content);
 420     }
 421
 422     my $query_params = decode_urlencoded($r->url->query());
 423
 424     foreach my $k (keys %{$query_params}) {
 425         $params->{$k} = $query_params->{$k};
 426     }
 427
 428     return PVE::Tools::decode_utf8_parameters($params);
 429 }
 430
 431 sub handle_api2_request {
 432     my ($self, $reqstate, $auth, $upload_state) = @_;
 433
 434     eval {
 435         my $r = $reqstate->{request};
 436         my $method = $r->method();
 437         my $path = $r->uri->path();
 438
 439         my ($rel_uri, $format) = split_abs_uri($path);
 440         if (!$format) {
 441             $self->error($reqstate, HTTP_NOT_IMPLEMENTED, "no such uri");
 442             return;
 443         }
 444
 445         #print Dumper($upload_state) if $upload_state;
 446
 447         my $rpcenv = $self->{rpcenv};
 448
 449         my $params;
 450
 451         if ($upload_state) {
 452             $params = $upload_state->{params};
 453         } else {
 454             $params = extract_params($r, $method);
 455         }
 456
 457         delete $params->{_dc}; # remove disable cache parameter
 458
 459         my $clientip = $reqstate->{peer_host};
 460
 461         $rpcenv->init_request();
 462
 463         my $res = PVE::REST::rest_handler($rpcenv, $clientip, $method, $rel_uri, $auth, $params);
 464
 465         AnyEvent->now_update(); # in case somebody called sleep()
 466
 467         $rpcenv->set_user(undef); # clear after request
 468
 469         if (my $host = $res->{proxy}) {
 470
 471             if ($self->{trusted_env}) {
 472                 $self->error($reqstate, HTTP_INTERNAL_SERVER_ERROR, "proxy not allowed");
 473                 return;
 474             }
 475
 476             if ($host ne 'localhost' && $r->header('PVEDisableProxy')) {
 477                 $self->error($reqstate, HTTP_INTERNAL_SERVER_ERROR, "proxy loop detected");
 478                 return;
 479             }
 480
 481             $res->{proxy_params}->{tmpfilename} = $reqstate->{tmpfilename} if $upload_state;
 482
 483             $self->proxy_request($reqstate, $clientip, $host, $method,
 484                                  $r->uri, $auth->{ticket}, $auth->{token}, $res->{proxy_params});
 485             return;
 486
 487         }
 488
 489         PVE::REST::prepare_response_data($format, $res);
 490         my ($raw, $ct, $nocomp) = PVE::REST::format_response_data($format, $res, $path);
 491
 492         my $resp = HTTP::Response->new($res->{status}, $res->{message});
 493         $resp->header("Content-Type" => $ct);
 494         $resp->content($raw);
 495         $self->response($reqstate, $resp, $nocomp);
 496     };
 497     if (my $err = $@) {
 498         $self->error($reqstate, 501, $err);
 499     }
 500 }
 501
 502 sub handle_spice_proxy_request {
 503     my ($self, $reqstate, $connect_str, $vmid, $node, $spiceport) = @_;
 504
 505     eval {
 506
 507         die "Port $spiceport is not allowed" if ($spiceport < 61000 || $spiceport > 61099);
 508
 509         my $rpcenv = $self->{rpcenv};
 510         $rpcenv->init_request();
 511
 512         my $clientip = $reqstate->{peer_host};
 513         my $r = $reqstate->{request};
 514
 515         my $remip;
 516
 517         if ($node ne 'localhost' && $node ne PVE::INotify::nodename()) {
 518             $remip = PVE::Cluster::remote_node_ip($node);
 519             die "unable to get remote IP address for node '$node'\n" if !$remip;
 520             print "REMOTE CONNECT $vmid, $remip, $connect_str\n" if $self->{debug};
 521         } else {
 522             print "$$: CONNECT $vmid, $node, $spiceport\n" if $self->{debug};
 523         }
 524
 525         if ($remip && $r->header('PVEDisableProxy')) {
 526             $self->error($reqstate, HTTP_INTERNAL_SERVER_ERROR, "proxy loop detected");
 527             return;
 528         }
 529
 530         $reqstate->{hdl}->timeout(0);
 531         $reqstate->{hdl}->wbuf_max(64*10*1024);
 532
 533         my $remhost = $remip ? $remip : "127.0.0.1";
 534         my $remport = $remip ? 3128 : $spiceport;
 535
 536         tcp_connect $remhost, $remport, sub {
 537             my ($fh) = @_
 538                 or die "connect to '$remhost:$remport' failed: $!";
 539
 540             print "$$: CONNECTed to '$remhost:$remport'\n" if $self->{debug};
 541             $reqstate->{proxyhdl} = AnyEvent::Handle->new(
 542                 fh => $fh,
 543                 rbuf_max => 64*1024,
 544                 wbuf_max => 64*10*1024,
 545                 timeout => 5,
 546                 on_eof => sub {
 547                     my ($hdl) = @_;
 548                     eval {
 549                         $self->log_aborted_request($reqstate);
 550                         $self->client_do_disconnect($reqstate);
 551                     };
 552                     if (my $err = $@) { syslog('err', $err); }
 553                 },
 554                 on_error => sub {
 555                     my ($hdl, $fatal, $message) = @_;
 556                     eval {
 557                         $self->log_aborted_request($reqstate, $message);
 558                         $self->client_do_disconnect($reqstate);
 559                     };
 560                     if (my $err = $@) { syslog('err', "$err"); }
 561                 });
 562
 563
 564             my $proxyhdlreader = sub {
 565                 my ($hdl) = @_;
 566
 567                 my $len = length($hdl->{rbuf});
 568                 my $data = substr($hdl->{rbuf}, 0, $len, '');
 569
 570                 #print "READ1 $len\n";
 571                 $reqstate->{hdl}->push_write($data) if $reqstate->{hdl};
 572             };
 573
 574             my $hdlreader = sub {
 575                 my ($hdl) = @_;
 576
 577                 my $len = length($hdl->{rbuf});
 578                 my $data = substr($hdl->{rbuf}, 0, $len, '');
 579
 580                 #print "READ0 $len\n";
 581                 $reqstate->{proxyhdl}->push_write($data) if $reqstate->{proxyhdl};
 582             };
 583
 584             my $proto = $reqstate->{proto} ? $reqstate->{proto}->{str} : 'HTTP/1.0';
 585
 586             my $startproxy = sub {
 587                 $reqstate->{proxyhdl}->timeout(0);
 588                 $reqstate->{proxyhdl}->on_read($proxyhdlreader);
 589                 $reqstate->{hdl}->on_read($hdlreader);
 590
 591                 # todo: use stop_read/start_read if write buffer grows to much
 592
 593                 my $res = "$proto 200 OK\015\012"; # hope this is the right answer?
 594                 $reqstate->{hdl}->push_write($res);
 595
 596                 # log early
 597                 $reqstate->{log}->{code} = 200;
 598                 $self->log_request($reqstate);
 599             };
 600
 601             if ($remip) {
 602                 my $header = "CONNECT ${connect_str} $proto\015\012" .
 603                     "Host: ${connect_str}\015\012" .
 604                     "Proxy-Connection: keep-alive\015\012" .
 605                     "User-Agent: spiceproxy\015\012" .
 606                     "PVEDisableProxy: true\015\012" .
 607                     "PVEClientIP: $clientip\015\012" .
 608                     "\015\012";
 609
 610                 $reqstate->{proxyhdl}->push_write($header);
 611                 $reqstate->{proxyhdl}->push_read(line => sub {
 612                     my ($hdl, $line) = @_;
 613
 614                     if ($line =~ m!^$proto 200 OK$!) {
 615                         &$startproxy();
 616                     } else {
 617                         $reqstate->{hdl}->push_write($line);
 618                         $self->client_do_disconnect($reqstate);
 619                     }
 620                 });
 621             } else {
 622                 &$startproxy();
 623             }
 624
 625         };
 626     };
 627     if (my $err = $@) {
 628         warn $err;
 629         $self->log_aborted_request($reqstate, $err);
 630         $self->client_do_disconnect($reqstate);
 631     }
 632 }
 633
 634 sub handle_request {
 635     my ($self, $reqstate, $auth) = @_;
 636
 637     eval {
 638         my $r = $reqstate->{request};
 639         my $method = $r->method();
 640         my $path = $r->uri->path();
 641
 642         # disable timeout on handle (we already have all data we need)
 643         # we re-enable timeout in response()
 644         $reqstate->{hdl}->timeout(0);
 645
 646         if ($path =~ m!$baseuri!) {
 647             $self->handle_api2_request($reqstate, $auth);
 648             return;
 649         }
 650
 651         if ($self->{pages} && ($method eq 'GET') && (my $handler = $self->{pages}->{$path})) {
 652             if (ref($handler) eq 'CODE') {
 653                 my $params = decode_urlencoded($r->url->query());
 654                 my ($resp, $userid) = &$handler($self, $reqstate->{request}, $params);
 655                 $self->response($reqstate, $resp);
 656             } elsif (ref($handler) eq 'HASH') {
 657                 if (my $filename = $handler->{file}) {
 658                     my $fh = IO::File->new($filename) ||
 659                         die "unable to open file '$filename' - $!\n";
 660                     send_file_start($self, $reqstate, $filename);
 661                 } else {
 662                     die "internal error - no handler";
 663                 }
 664             } else {
 665                 die "internal error - no handler";
 666             }
 667             return;
 668         }
 669
 670         if ($self->{dirs} && ($method eq 'GET')) {
 671             # we only allow simple names
 672             if ($path =~ m!^(/\S+/)([a-zA-Z0-9\-\_\.]+)$!) {
 673                 my ($subdir, $file) = ($1, $2);
 674                 if (my $dir = $self->{dirs}->{$subdir}) {
 675                     my $filename = "$dir$file";
 676                     my $fh = IO::File->new($filename) ||
 677                         die "unable to open file '$filename' - $!\n";
 678                     send_file_start($self, $reqstate, $filename);
 679                     return;
 680                 }
 681             }
 682         }
 683
 684         die "no such file '$path'";
 685     };
 686     if (my $err = $@) {
 687         $self->error($reqstate, 501, $err);
 688     }
 689 }
 690
 691 sub file_upload_multipart {
 692     my ($self, $reqstate, $auth, $rstate) = @_;
 693
 694     eval {
 695         my $boundary = $rstate->{boundary};
 696         my $hdl = $reqstate->{hdl};
 697
 698         my $startlen = length($hdl->{rbuf});
 699
 700         if ($rstate->{phase} == 0) { # skip everything until start
 701             if ($hdl->{rbuf} =~ s/^.*?--\Q$boundary\E  \015?\012
 702                        ((?:[^\015]+\015\012)* ) \015?\012//xs) {
 703                 my $header = $1;
 704                 my ($ct, $disp, $name, $filename);
 705                 foreach my $line (split(/\015?\012/, $header)) {
 706                     # assume we have single line headers
 707                     if ($line =~ m/^Content-Type\s*:\s*(.*)/i) {
 708                         $ct = parse_content_type($1);
 709                     } elsif ($line =~ m/^Content-Disposition\s*:\s*(.*)/i) {
 710                         ($disp, $name, $filename) = parse_content_disposition($1);
 711                     }
 712                 }
 713
 714                 if (!($disp && $disp eq 'form-data' && $name)) {
 715                     syslog('err', "wrong content disposition in multipart - abort upload");
 716                     $rstate->{phase} = -1;
 717                 } else {
 718
 719                     $rstate->{fieldname} = $name;
 720
 721                     if ($filename) {
 722                         if ($name eq 'filename') {
 723                             # found file upload data
 724                             $rstate->{phase} = 1;
 725                             $rstate->{filename} = $filename;
 726                         } else {
 727                             syslog('err', "wrong field name for file upload - abort upload");
 728                             $rstate->{phase} = -1;
 729                         }
 730                     } else {
 731                         # found form data for field $name
 732                         $rstate->{phase} = 2;
 733                     }
 734                 }
 735             } else {
 736                 my $len = length($hdl->{rbuf});
 737                 substr($hdl->{rbuf}, 0, $len - $rstate->{maxheader}, '')
 738                     if $len > $rstate->{maxheader}; # skip garbage
 739             }
 740         } elsif ($rstate->{phase} == 1) { # inside file - dump until end marker
 741             if ($hdl->{rbuf} =~ s/^(.*?)\015?\012(--\Q$boundary\E(--)? \015?\012(.*))$/$2/xs) {
 742                 my ($rest, $eof) = ($1, $3);
 743                 my $len = length($rest);
 744                 die "write to temporary file failed - $!"
 745                     if syswrite($rstate->{outfh}, $rest) != $len;
 746                 $rstate->{ctx}->add($rest);
 747                 $rstate->{params}->{filename} = $rstate->{filename};
 748                 $rstate->{md5sum} = $rstate->{ctx}->hexdigest;
 749                 $rstate->{bytes} += $len;
 750                 $rstate->{phase} =  $eof ? 100 : 0;
 751             } else {
 752                 my $len = length($hdl->{rbuf});
 753                 my $wlen = $len - $rstate->{boundlen};
 754                 if ($wlen > 0) {
 755                     my $data = substr($hdl->{rbuf}, 0, $wlen, '');
 756                     die "write to temporary file failed - $!"
 757                         if syswrite($rstate->{outfh}, $data) != $wlen;
 758                     $rstate->{bytes} += $wlen;
 759                     $rstate->{ctx}->add($data);
 760                 }
 761             }
 762         } elsif ($rstate->{phase} == 2) { # inside normal field
 763
 764             if ($hdl->{rbuf} =~ s/^(.*?)\015?\012(--\Q$boundary\E(--)? \015?\012(.*))$/$2/xs) {
 765                 my ($rest, $eof) = ($1, $3);
 766                 my $len = length($rest);
 767                 $rstate->{post_size} += $len;
 768                 if ($rstate->{post_size} < $limit_max_post) {
 769                     $rstate->{params}->{$rstate->{fieldname}} = $rest;
 770                     $rstate->{phase} = $eof ? 100 : 0;
 771                 } else {
 772                     syslog('err', "form data to large - abort upload");
 773                     $rstate->{phase} = -1; # skip
 774                 }
 775             }
 776         } else { # skip
 777             my $len = length($hdl->{rbuf});
 778             substr($hdl->{rbuf}, 0, $len, ''); # empty rbuf
 779         }
 780
 781         $rstate->{read} += ($startlen - length($hdl->{rbuf}));
 782
 783         if (!$rstate->{done} && ($rstate->{read} + length($hdl->{rbuf})) >= $rstate->{size}) {
 784             $rstate->{done} = 1; # make sure we dont get called twice
 785             if ($rstate->{phase} < 0 || !$rstate->{md5sum}) {
 786                 die "upload failed\n";
 787             } else {
 788                 my $elapsed = tv_interval($rstate->{starttime});
 789
 790                 my $rate = int($rstate->{bytes}/($elapsed*1024*1024));
 791                 syslog('info', "multipart upload complete " .
 792                        "(size: %d time: %ds rate: %.2fMiB/s md5sum: $rstate->{md5sum})",
 793                        $rstate->{bytes}, $elapsed, $rate);
 794                 $self->handle_api2_request($reqstate, $auth, $rstate);
 795             }
 796         }
 797     };
 798     if (my $err = $@) {
 799         syslog('err', $err);
 800         $self->error($reqstate, 501, $err);
 801     }
 802 }
 803
 804 sub parse_content_type {
 805     my ($ctype) = @_;
 806
 807     my ($ct, @params) = split(/\s*[;,]\s*/o, $ctype);
 808
 809     foreach my $v (@params) {
 810         if ($v =~ m/^\s*boundary\s*=\s*(\S+?)\s*$/o) {
 811             return wantarray ? ($ct, $1) : $ct;
 812         }
 813     }
 814
 815     return  wantarray ? ($ct) : $ct;
 816 }
 817
 818 sub parse_content_disposition {
 819     my ($line) = @_;
 820
 821     my ($disp, @params) = split(/\s*[;,]\s*/o, $line);
 822     my $name;
 823     my $filename;
 824
 825     foreach my $v (@params) {
 826         if ($v =~ m/^\s*name\s*=\s*(\S+?)\s*$/o) {
 827             $name = $1;
 828             $name =~ s/^"(.*)"$/$1/;
 829         } elsif ($v =~ m/^\s*filename\s*=\s*(.+?)\s*$/o) {
 830             $filename = $1;
 831             $filename =~ s/^"(.*)"$/$1/;
 832         }
 833     }
 834
 835     return  wantarray ? ($disp, $name, $filename) : $disp;
 836 }
 837
 838 my $tmpfile_seq_no = 0;
 839
 840 sub get_upload_filename {
 841     # choose unpredictable tmpfile name
 842
 843     $tmpfile_seq_no++;
 844     return "/var/tmp/pveupload-" . Digest::MD5::md5_hex($tmpfile_seq_no . time() . $$);
 845 }
 846
 847 sub unshift_read_header {
 848     my ($self, $reqstate, $state) = @_;
 849
 850     $state = { size => 0, count => 0 } if !$state;
 851
 852     $reqstate->{hdl}->unshift_read(line => sub {
 853         my ($hdl, $line) = @_;
 854
 855         eval {
 856             # print "$$: got header: $line\n" if $self->{debug};
 857
 858             die "to many http header lines\n" if ++$state->{count} >= $limit_max_headers;
 859             die "http header too large\n" if ($state->{size} += length($line)) >= $limit_max_header_size;
 860
 861             my $r = $reqstate->{request};
 862             if ($line eq '') {
 863
 864                 my $path = $r->uri->path();
 865                 my $method = $r->method();
 866
 867                 $r->push_header($state->{key}, $state->{val})
 868                     if $state->{key};
 869
 870                 if (!$known_methods->{$method}) {
 871                     my $resp = HTTP::Response->new(HTTP_NOT_IMPLEMENTED, "method '$method' not available");
 872                     $self->response($reqstate, $resp);
 873                     return;
 874                 }
 875
 876                 my $conn = $r->header('Connection');
 877                 my $accept_enc = $r->header('Accept-Encoding');
 878                 $reqstate->{accept_gzip} = ($accept_enc && $accept_enc =~ m/gzip/) ? 1 : 0;
 879
 880                 if ($conn) {
 881                     $reqstate->{keep_alive} = 0 if $conn =~ m/close/oi;
 882                 } else {
 883                     if ($reqstate->{proto}->{ver} < 1001) {
 884                         $reqstate->{keep_alive} = 0;
 885                     }
 886                 }
 887
 888                 my $te  = $r->header('Transfer-Encoding');
 889                 if ($te && lc($te) eq 'chunked') {
 890                     # Handle chunked transfer encoding
 891                     $self->error($reqstate, 501, "chunked transfer encoding not supported");
 892                     return;
 893                 } elsif ($te) {
 894                     $self->error($reqstate, 501, "Unknown transfer encoding '$te'");
 895                     return;
 896                 }
 897
 898                 my $pveclientip = $r->header('PVEClientIP');
 899
 900                 # fixme: how can we make PVEClientIP header trusted?
 901                 if ($self->{trusted_env} && $pveclientip) {
 902                     $reqstate->{peer_host} = $pveclientip;
 903                 } else {
 904                     $r->header('PVEClientIP', $reqstate->{peer_host});
 905                 }
 906
 907                 my $len = $r->header('Content-Length');
 908
 909                 # header processing complete - authenticate now
 910
 911                 my $auth = {};
 912                 if ($self->{spiceproxy}) {
 913                     my $connect_str = $r->header('Host');
 914                     my ($vmid, $node, $port) = PVE::AccessControl::verify_spice_connect_url($connect_str);
 915                     if (!($vmid && $node && $port)) {
 916                         $self->error($reqstate, HTTP_UNAUTHORIZED, "invalid ticket");
 917                         return;
 918                     }
 919                     $self->handle_spice_proxy_request($reqstate, $connect_str, $vmid, $node, $port);
 920                     return;
 921                 } elsif ($path =~ m!$baseuri!) {
 922                     my $token = $r->header('CSRFPreventionToken');
 923                     my $cookie = $r->header('Cookie');
 924                     my $ticket = PVE::REST::extract_auth_cookie($cookie);
 925
 926                     my ($rel_uri, $format) = split_abs_uri($path);
 927                     if (!$format) {
 928                         $self->error($reqstate, HTTP_NOT_IMPLEMENTED, "no such uri");
 929                         return;
 930                     }
 931
 932                     eval {
 933                         $auth = PVE::REST::auth_handler($self->{rpcenv}, $reqstate->{peer_host}, $method,
 934                                                         $rel_uri, $ticket, $token);
 935                     };
 936                     if (my $err = $@) {
 937                         $self->error($reqstate, HTTP_UNAUTHORIZED, $err);
 938                         return;
 939                     }
 940                 }
 941
 942                 $reqstate->{log}->{userid} = $auth->{userid};
 943
 944                 if ($len) {
 945
 946                     if (!($method eq 'PUT' || $method eq 'POST')) {
 947                         $self->error($reqstate, 501, "Unexpected content for method '$method'");
 948                         return;
 949                     }
 950
 951                     my $ctype = $r->header('Content-Type');
 952                     my ($ct, $boundary) = parse_content_type($ctype) if $ctype;
 953
 954                     if ($auth->{isUpload} && !$self->{trusted_env}) {
 955                         die "upload 'Content-Type '$ctype' not implemented\n"
 956                             if !($boundary && $ct && ($ct eq 'multipart/form-data'));
 957
 958                         die "upload without content length header not supported" if !$len;
 959
 960                         die "upload without content length header not supported" if !$len;
 961
 962                         print "start upload $path $ct $boundary\n" if $self->{debug};
 963
 964                         my $tmpfilename = get_upload_filename();
 965                         my $outfh = IO::File->new($tmpfilename, O_RDWR|O_CREAT|O_EXCL, 0600) ||
 966                             die "unable to create temporary upload file '$tmpfilename'";
 967
 968                         $reqstate->{keep_alive} = 0;
 969
 970                         my $boundlen = length($boundary) + 8; # \015?\012--$boundary--\015?\012
 971
 972                         my $state = {
 973                             size => $len,
 974                             boundary => $boundary,
 975                             ctx => Digest::MD5->new,
 976                             boundlen =>  $boundlen,
 977                             maxheader => 2048 + $boundlen, # should be large enough
 978                             params => decode_urlencoded($r->url->query()),
 979                             phase => 0,
 980                             read => 0,
 981                             post_size => 0,
 982                             starttime => [gettimeofday],
 983                             outfh => $outfh,
 984                         };
 985                         $reqstate->{tmpfilename} = $tmpfilename;
 986                         $reqstate->{hdl}->on_read(sub { $self->file_upload_multipart($reqstate, $auth, $state); });
 987                         return;
 988                     }
 989
 990                     if ($len > $limit_max_post) {
 991                         $self->error($reqstate, 501, "for data too large");
 992                         return;
 993                     }
 994
 995                     if (!$ct || $ct eq 'application/x-www-form-urlencoded') {
 996                         $reqstate->{hdl}->unshift_read(chunk => $len, sub {
 997                             my ($hdl, $data) = @_;
 998                             $r->content($data);
 999                             $self->handle_request($reqstate, $auth);
1000                         });
1001                     } else {
1002                         $self->error($reqstate, 506, "upload 'Content-Type '$ctype' not implemented");
1003                     }
1004                 } else {
1005                     $self->handle_request($reqstate, $auth);
1006                 }
1007             } elsif ($line =~ /^([^:\s]+)\s*:\s*(.*)/) {
1008                 $r->push_header($state->{key}, $state->{val}) if $state->{key};
1009                 ($state->{key}, $state->{val}) = ($1, $2);
1010                 $self->unshift_read_header($reqstate, $state);
1011             } elsif ($line =~ /^\s+(.*)/) {
1012                 $state->{val} .= " $1";
1013                 $self->unshift_read_header($reqstate, $state);
1014             } else {
1015                 $self->error($reqstate, 506, "unable to parse request header");
1016             }
1017         };
1018         warn $@ if $@;
1019     });
1020 };
1021
1022 sub push_request_header {
1023     my ($self, $reqstate) = @_;
1024
1025     eval {
1026         $reqstate->{hdl}->push_read(line => sub {
1027             my ($hdl, $line) = @_;
1028
1029             eval {
1030                 # print "got request header: $line\n" if $self->{debug};
1031
1032                 $reqstate->{keep_alive}--;
1033
1034                 if ($line =~ /(\S+)\040(\S+)\040HTTP\/(\d+)\.(\d+)/o) {
1035                     my ($method, $uri, $maj, $min) = ($1, $2, $3, $4);
1036
1037                     if ($maj != 1) {
1038                         $self->error($reqstate, 506, "http protocol version $maj.$min not supported");
1039                         return;
1040                     }
1041
1042                     $self->{request_count}++; # only count valid request headers
1043                     if ($self->{request_count} >= $self->{max_requests}) {
1044                         $self->{end_loop} = 1;
1045                     }
1046                     $reqstate->{log} = { requestline => $line };
1047                     $reqstate->{proto}->{str} = "HTTP/$maj.$min";
1048                     $reqstate->{proto}->{maj} = $maj;
1049                     $reqstate->{proto}->{min} = $min;
1050                     $reqstate->{proto}->{ver} = $maj*1000+$min;
1051                     $reqstate->{request} = HTTP::Request->new($method, $uri);
1052
1053                     $self->unshift_read_header($reqstate);
1054                 } elsif ($line eq '') {
1055                     # ignore empty lines before requests (browser bugs?)
1056                     $self->push_request_header($reqstate);
1057                 } else {
1058                     $self->error($reqstate, 400, 'bad request');
1059                 }
1060             };
1061             warn $@ if $@;
1062         });
1063     };
1064     warn $@ if $@;
1065 }
1066
1067 sub accept {
1068     my ($self) = @_;
1069
1070     my $clientfh;
1071
1072     return if $self->{end_loop};
1073
1074     # we need to m make sure that only one process calls accept
1075     while (!flock($self->{lockfh}, Fcntl::LOCK_EX())) {
1076         next if $! == EINTR;
1077         die "could not get lock on file '$self->{lockfile}' -  $!\n";
1078     }
1079
1080     my $again = 0;
1081     my $errmsg;
1082     eval {
1083         while (!$self->{end_loop} &&
1084                !defined($clientfh = $self->{socket}->accept()) &&
1085                ($! == EINTR)) {};
1086
1087         if ($self->{end_loop}) {
1088             $again = 0;
1089         } else {
1090             $again = ($! == EAGAIN || $! == WSAEWOULDBLOCK);
1091             if (!defined($clientfh)) {
1092                 $errmsg = "failed to accept connection: $!\n";
1093             }
1094         }
1095     };
1096     warn $@ if $@;
1097
1098     flock($self->{lockfh}, Fcntl::LOCK_UN());
1099
1100     if (!defined($clientfh)) {
1101         return if $again;
1102         die $errmsg if $errmsg;
1103     }
1104
1105     fh_nonblocking $clientfh, 1;
1106
1107     $self->{conn_count}++;
1108
1109     return $clientfh;
1110 }
1111
1112 sub wait_end_loop {
1113     my ($self) = @_;
1114
1115     $self->{end_loop} = 1;
1116
1117     undef $self->{socket_watch};
1118
1119     if ($self->{conn_count} <= 0) {
1120         $self->{end_cond}->send(1);
1121         return;
1122     }
1123
1124     # else we need to wait until all open connections gets closed
1125     my $w; $w = AnyEvent->timer (after => 1, interval => 1, cb => sub {
1126         eval {
1127             # todo: test for active connections instead (we can abort idle connections)
1128             if ($self->{conn_count} <= 0) {
1129                 undef $w;
1130                 $self->{end_cond}->send(1);
1131             }
1132         };
1133         warn $@ if $@;
1134     });
1135 }
1136
1137
1138 sub check_host_access {
1139     my ($self, $clientip) = @_;
1140
1141     my $cip = Net::IP->new($clientip);
1142
1143     my $match_allow = 0;
1144     my $match_deny = 0;
1145
1146     if ($self->{allow_from}) {
1147         foreach my $t (@{$self->{allow_from}}) {
1148             if ($t->overlaps($cip)) {
1149                 $match_allow = 1;
1150                 last;
1151             }
1152         }
1153     }
1154
1155     if ($self->{deny_from}) {
1156         foreach my $t (@{$self->{deny_from}}) {
1157             if ($t->overlaps($cip)) {
1158                 $match_deny = 1;
1159                 last;
1160             }
1161         }
1162     }
1163
1164     if ($match_allow == $match_deny) {
1165         # match both allow and deny, or no match
1166         return $self->{policy} && $self->{policy} eq 'allow' ? 1 : 0;
1167     }
1168
1169     return $match_allow;
1170 }
1171
1172 sub accept_connections {
1173     my ($self) = @_;
1174
1175     eval {
1176
1177         while (my $clientfh = $self->accept()) {
1178
1179             my $reqstate = { keep_alive => $self->{keep_alive} };
1180
1181             # stop keep-alive when there are many open connections
1182             if ($self->{conn_count} >= $self->{max_conn_soft_limit}) {
1183                 $reqstate->{keep_alive} = 0;
1184             }
1185
1186             if (my $sin = getpeername($clientfh)) {
1187                 my ($pport, $phost) = Socket::unpack_sockaddr_in($sin);
1188                 ($reqstate->{peer_port}, $reqstate->{peer_host}) = ($pport,  Socket::inet_ntoa($phost));
1189             }
1190
1191             if (!$self->{trusted_env} && !$self->check_host_access($reqstate->{peer_host})) {
1192                 print "$$: ABORT request from $reqstate->{peer_host} - access denied\n" if $self->{debug};
1193                 $reqstate->{log}->{code} = 403;
1194                 $self->log_request($reqstate);
1195                 next;
1196             }
1197
1198             $reqstate->{hdl} = AnyEvent::Handle->new(
1199                 fh => $clientfh,
1200                 rbuf_max => 64*1024,
1201                 timeout => $self->{timeout},
1202                 linger => 0, # avoid problems with ssh - really needed ?
1203                 on_eof   => sub {
1204                     my ($hdl) = @_;
1205                     eval {
1206                         $self->log_aborted_request($reqstate);
1207                         $self->client_do_disconnect($reqstate);
1208                     };
1209                     if (my $err = $@) { syslog('err', $err); }
1210                 },
1211                 on_error => sub {
1212                     my ($hdl, $fatal, $message) = @_;
1213                     eval {
1214                         $self->log_aborted_request($reqstate, $message);
1215                         $self->client_do_disconnect($reqstate);
1216                     };
1217                     if (my $err = $@) { syslog('err', "$err"); }
1218                 },
1219                 ($self->{tls_ctx} ? (tls => "accept", tls_ctx => $self->{tls_ctx}) : ()));
1220
1221             print "$$: ACCEPT FH" .  $clientfh->fileno() . " CONN$self->{conn_count}\n" if $self->{debug};
1222
1223             $self->push_request_header($reqstate);
1224         }
1225     };
1226
1227     if (my $err = $@) {
1228         syslog('err', $err);
1229         $self->{end_loop} = 1;
1230     }
1231
1232     $self->wait_end_loop() if $self->{end_loop};
1233 }
1234
1235 # Note: We can't open log file in non-blocking mode and use AnyEvent::Handle,
1236 # because we write from multiple processes, and that would arbitrarily mix output
1237 # of all processes.
1238 sub open_access_log {
1239     my ($self, $filename) = @_;
1240
1241     my $old_mask = umask(0137);;
1242     my $logfh = IO::File->new($filename, ">>") ||
1243         die "unable to open log file '$filename' - $!\n";
1244     umask($old_mask);
1245
1246     $logfh->autoflush(1);
1247
1248     $self->{logfh} = $logfh;
1249 }
1250
1251 sub write_log {
1252     my ($self, $data) = @_;
1253
1254     return if !defined($self->{logfh}) || !$data;
1255
1256     my $res = $self->{logfh}->print($data);
1257
1258     if (!$res) {
1259         delete $self->{logfh};
1260         syslog('err', "error writing access log");
1261         $self->{end_loop} = 1; # terminate asap
1262     }
1263 }
1264
1265 sub atfork_handler {
1266     my ($self) = @_;
1267
1268     eval {
1269         # something else do to ?
1270         close($self->{socket});
1271     };
1272     warn $@ if $@;
1273 }
1274
1275 sub new {
1276     my ($this, %args) = @_;
1277
1278     my $class = ref($this) || $this;
1279
1280     foreach my $req (qw(socket lockfh lockfile)) {
1281         die "misssing required argument '$req'" if !defined($args{$req});
1282     }
1283
1284     my $self = bless { %args }, $class;
1285
1286     # init inotify
1287     PVE::INotify::inotify_init();
1288
1289     $self->{rpcenv} = PVE::RPCEnvironment->init(
1290         $self->{trusted_env} ? 'priv' : 'pub', atfork =>  sub { $self-> atfork_handler() });
1291
1292     fh_nonblocking($self->{socket}, 1);
1293
1294     $self->{end_loop} = 0;
1295     $self->{conn_count} = 0;
1296     $self->{request_count} = 0;
1297     $self->{timeout} = 5 if !$self->{timeout};
1298     $self->{keep_alive} = 0 if !defined($self->{keep_alive});
1299     $self->{max_conn} = 800 if !$self->{max_conn};
1300     $self->{max_requests} = 8000 if !$self->{max_requests};
1301
1302     $self->{policy} = 'allow' if !$self->{policy};
1303
1304     $self->{end_cond} = AnyEvent->condvar;
1305
1306     if ($self->{ssl}) {
1307         $self->{tls_ctx} = AnyEvent::TLS->new(%{$self->{ssl}});
1308         Net::SSLeay::CTX_set_options($self->{tls_ctx}->{ctx}, &Net::SSLeay::OP_NO_COMPRESSION);
1309     }
1310
1311     if ($self->{spiceproxy}) {
1312         $known_methods = { CONNECT => 1 };
1313     }
1314
1315     $self->open_access_log($self->{logfile}) if $self->{logfile};
1316
1317     $self->{max_conn_soft_limit} = $self->{max_conn} > 100 ? $self->{max_conn} - 20 : $self->{max_conn};
1318
1319     $self->{socket_watch} = AnyEvent->io(fh => $self->{socket}, poll => 'r', cb => sub {
1320         eval {
1321             if ($self->{conn_count} >= $self->{max_conn}) {
1322                 my $w; $w = AnyEvent->timer (after => 1, interval => 1, cb => sub {
1323                     if ($self->{conn_count} < $self->{max_conn}) {
1324                         undef $w;
1325                         $self->accept_connections();
1326                     }
1327                 });
1328             } else {
1329                 $self->accept_connections();
1330             }
1331         };
1332         warn $@ if $@;
1333     });
1334
1335     $self->{term_watch} = AnyEvent->signal(signal => "TERM", cb => sub {
1336         undef $self->{term_watch};
1337         $self->wait_end_loop();
1338     });
1339
1340     $self->{quit_watch} = AnyEvent->signal(signal => "QUIT", cb => sub {
1341         undef $self->{quit_watch};
1342         $self->wait_end_loop();
1343     });
1344
1345     $self->{inotify_poll} = AnyEvent->timer(after => 5, interval => 5, cb => sub {
1346         PVE::INotify::poll(); # read inotify events
1347     });
1348
1349     return $self;
1350 }
1351
1352 sub run {
1353     my ($self) = @_;
1354
1355     $self->{end_cond}->recv;
1356 }
1357
1358 1;