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