PVE/RPCEnvironment.pm

   1 package PVE::RPCEnvironment;
   2
   3 use strict;
   4 use warnings;
   5 use POSIX qw(:sys_wait_h EINTR);
   6 use IO::Handle;
   7 use IO::File;
   8 use IO::Select;
   9 use Fcntl qw(:flock);
  10 use PVE::SafeSyslog;
  11 use PVE::Tools;
  12 use PVE::INotify;
  13 use PVE::Cluster;
  14 use PVE::ProcFSTools;
  15 use PVE::AccessControl;
  16
  17 # we use this singleton class to pass RPC related environment values
  18
  19 my $pve_env;
  20
  21 # save $SIG{CHLD} handler implementation.
  22 # simply set $SIG{CHLD} = $worker_reaper;
  23 # and register forked processes with &$register_worker(pid)
  24 # Note: using $SIG{CHLD} = 'IGNORE' or $SIG{CHLD} = sub { wait (); } or ...
  25 # has serious side effects, because perls built in system() and open()
  26 # functions can't get the correct exit status of a child. So we cant use
  27 # that (also see perlipc)
  28
  29 my $WORKER_PIDS;
  30
  31 my $log_task_result = sub {
  32     my ($upid, $user, $status) = @_;
  33
  34     my $msg = 'successful';
  35     my $pri = 'info';
  36     if ($status != 0) {
  37         my $ec = $status >> 8;
  38         my $ic = $status & 255;
  39         $msg = $ec ? "failed ($ec)" : "interrupted ($ic)";
  40         $pri = 'err';
  41     }
  42     my $tlist = active_workers($upid);
  43     PVE::Cluster::broadcast_tasklist($tlist);
  44     my $task;
  45     foreach my $t (@$tlist) {
  46         if ($t->{upid} eq $upid) {
  47             $task = $t;
  48             last;
  49         }
  50     }
  51     if ($task && $task->{status}) {
  52         $msg = $task->{status};
  53     }
  54     PVE::Cluster::log_msg($pri, $user, "end task $upid $msg");
  55 };
  56
  57 my $worker_reaper = sub {
  58     local $!; local $?;
  59     foreach my $pid (keys %$WORKER_PIDS) {
  60         my $waitpid = waitpid ($pid, WNOHANG);
  61         if (defined($waitpid) && ($waitpid == $pid)) {
  62             my $info = $WORKER_PIDS->{$pid};
  63             if ($info && $info->{upid} && $info->{user}) {
  64                 &$log_task_result($info->{upid}, $info->{user}, $?);
  65             }
  66             delete ($WORKER_PIDS->{$pid});
  67         }
  68     }
  69 };
  70
  71 my $register_worker = sub {
  72     my ($pid, $user, $upid) = @_;
  73
  74     return if !$pid;
  75
  76     # do not register if already finished
  77     my $waitpid = waitpid ($pid, WNOHANG);
  78     if (defined($waitpid) && ($waitpid == $pid)) {
  79         delete ($WORKER_PIDS->{$pid});
  80         return;
  81     }
  82
  83     $WORKER_PIDS->{$pid} = {
  84         user => $user,
  85         upid => $upid,
  86     };
  87 };
  88
  89 # ACL cache
  90
  91 my $compile_acl = sub {
  92     my ($self, $user) = @_;
  93
  94     my $res = {};
  95     my $cfg = $self->{user_cfg};
  96
  97     return undef if !$cfg->{roles};
  98
  99     if ($user eq 'root@pam') { # root can do anything
 100         return {'/' => $cfg->{roles}->{'Administrator'}};
 101     }
 102
 103     foreach my $path (sort keys %{$cfg->{acl}}) {
 104         my @ra = PVE::AccessControl::roles($cfg, $user, $path);
 105
 106         my $privs = {};
 107         foreach my $role (@ra) {
 108             if (my $privset = $cfg->{roles}->{$role}) {
 109                 foreach my $p (keys %$privset) {
 110                     $privs->{$p} = 1;
 111                 }
 112             }
 113         }
 114
 115         $res->{$path} = $privs;
 116     }
 117
 118     return $res;
 119 };
 120
 121 sub permissions {
 122     my ($self, $user, $path) = @_;
 123
 124     $user = PVE::AccessControl::verify_username($user, 1);
 125     return {} if !$user;
 126
 127     my $cache = $self->{aclcache};
 128
 129     my $acl = $cache->{$user};
 130
 131     if (!$acl) {
 132         if (!($acl = &$compile_acl($self, $user))) {
 133             return {};
 134         }
 135         $cache->{$user} = $acl;
 136     }
 137
 138     my $perm;
 139
 140     if (!($perm = $acl->{$path})) {
 141         $perm = {};
 142         foreach my $p (sort keys %$acl) {
 143             my $final = ($path eq $p);
 144
 145             next if !(($p eq '/') || $final || ($path =~ m|^$p/|));
 146
 147             $perm = $acl->{$p};
 148         }
 149         $acl->{$path} = $perm;
 150     }
 151
 152     return $perm;
 153 }
 154
 155 sub check {
 156     my ($self, $user, $path, $privs) = @_;
 157
 158     my $perm = $self->permissions($user, $path);
 159
 160     foreach my $priv (@$privs) {
 161         return undef if !$perm->{$priv};
 162     };
 163
 164     return 1;
 165 };
 166
 167 sub user_enabled {
 168     my ($self, $user) = @_;
 169
 170     my $cfg = $self->{user_cfg};
 171     return PVE::AccessControl::user_enabled($cfg, $user);
 172 }
 173
 174 # initialize environment - must be called once at program startup
 175 sub init {
 176     my ($class, $type, %params) = @_;
 177
 178     $class = ref($class) || $class;
 179
 180     die "already initialized" if $pve_env;
 181
 182     die "unknown environment type" if !$type || $type !~ m/^(cli|pub|priv)$/;
 183
 184     $SIG{CHLD} = $worker_reaper;
 185
 186     # environment types
 187     # cli  ... command started fron command line
 188     # pub  ... access from public server (apache)
 189     # priv ... access from private server (pvedaemon)
 190
 191     my $self = {
 192         user_cfg => {},
 193         aclcache => {},
 194         aclversion => undef,
 195         type => $type,
 196     };
 197
 198     bless $self, $class;
 199
 200     foreach my $p (keys %params) {
 201         if ($p eq 'atfork') {
 202             $self->{$p} = $params{$p};
 203         } else {
 204             die "unknown option '$p'";
 205         }
 206     }
 207
 208     $pve_env = $self;
 209
 210     my ($sysname, $nodename) = POSIX::uname();
 211
 212     $nodename =~ s/\..*$//; # strip domain part, if any
 213
 214     $self->{nodename} = $nodename;
 215
 216     return $self;
 217 };
 218
 219 # get the singleton
 220 sub get {
 221
 222     die "not initialized" if !$pve_env;
 223
 224     return $pve_env;
 225 }
 226
 227 # init_request - must be called before each RPC request
 228 sub init_request {
 229     my ($self, %params) = @_;
 230
 231     PVE::Cluster::cfs_update();
 232
 233     my $userconfig; # we use this for regression tests
 234     foreach my $p (keys %params) {
 235         if ($p eq 'userconfig') {
 236             $userconfig = $params{$p};
 237         } else {
 238             die "unknown parameter '$p'";
 239         }
 240     }
 241
 242     eval {
 243         $self->{aclcache} = {};
 244         if ($userconfig) {
 245             my $ucdata = PVE::Tools::file_get_contents($userconfig);
 246             my $cfg = PVE::AccessControl::parse_user_config($userconfig, $ucdata);
 247             $self->{user_cfg} = $cfg;
 248         } else {
 249             my $ucvers = PVE::Cluster::cfs_file_version('user.cfg');
 250             if (!$self->{aclcache} || !defined($self->{aclversion}) ||
 251                 !defined($ucvers) ||  ($ucvers ne $self->{aclversion})) {
 252                 $self->{aclversion} = $ucvers;
 253                 my $cfg = PVE::Cluster::cfs_read_file('user.cfg');
 254                 $self->{user_cfg} = $cfg;
 255             }
 256         }
 257     };
 258     if (my $err = $@) {
 259         $self->{user_cfg} = {};
 260         die "Unable to load access control list: $err";
 261     }
 262 }
 263
 264 sub set_client_ip {
 265     my ($self, $ip) = @_;
 266
 267     $self->{client_ip} = $ip;
 268 }
 269
 270 sub get_client_ip {
 271     my ($self) = @_;
 272
 273     return $self->{client_ip};
 274 }
 275
 276 sub set_result_count {
 277     my ($self, $count) = @_;
 278
 279     $self->{result_count} = $count;
 280 }
 281
 282 sub get_result_count {
 283     my ($self) = @_;
 284
 285     return $self->{result_count};
 286 }
 287
 288 sub set_language {
 289     my ($self, $lang) = @_;
 290
 291     # fixme: initialize I18N
 292
 293     $self->{language} = $lang;
 294 }
 295
 296 sub get_language {
 297     my ($self) = @_;
 298
 299     return $self->{language};
 300 }
 301
 302 sub set_user {
 303     my ($self, $user) = @_;
 304
 305     # fixme: get ACLs
 306
 307     $self->{user} = $user;
 308 }
 309
 310 sub get_user {
 311     my ($self) = @_;
 312
 313     die "user name not set\n" if !$self->{user};
 314
 315     return $self->{user};
 316 }
 317
 318 # read/update list of active workers
 319 # we move all finished tasks to the archive index,
 320 # but keep aktive and most recent task in the active file.
 321 # $nocheck ... consider $new_upid still running (avoid that
 322 # we try to read the reult to early.
 323 sub active_workers  {
 324     my ($new_upid, $nocheck) = @_;
 325
 326     my $lkfn = "/var/log/pve/tasks/.active.lock";
 327
 328     my $timeout = 10;
 329
 330     my $code = sub {
 331
 332         my $tasklist = PVE::INotify::read_file('active');
 333
 334         my @ta;
 335         my $tlist = [];
 336         my $thash = {}; # only list task once
 337
 338         my $check_task = sub {
 339             my ($task, $running) = @_;
 340
 341             if ($running || PVE::ProcFSTools::check_process_running($task->{pid}, $task->{pstart})) {
 342                 push @$tlist, $task;
 343             } else {
 344                 delete $task->{pid};
 345                 push @ta, $task;
 346             }
 347             delete $task->{pstart};
 348         };
 349
 350         foreach my $task (@$tasklist) {
 351             my $upid = $task->{upid};
 352             next if $thash->{$upid};
 353             $thash->{$upid} = $task;
 354             &$check_task($task);
 355         }
 356
 357         if ($new_upid && !(my $task = $thash->{$new_upid})) {
 358             $task = PVE::Tools::upid_decode($new_upid);
 359             $task->{upid} = $new_upid;
 360             $thash->{$new_upid} = $task;
 361             &$check_task($task, $nocheck);
 362         }
 363
 364
 365         @ta = sort { $b->{starttime} cmp $a->{starttime} } @ta;
 366
 367         my $save = defined($new_upid);
 368
 369         foreach my $task (@ta) {
 370             next if $task->{endtime};
 371             $task->{endtime} = time();
 372             $task->{status} = PVE::Tools::upid_read_status($task->{upid});
 373             $save = 1;
 374         }
 375
 376         my $archive = '';
 377         my @arlist = ();
 378         foreach my $task (@ta) {
 379             if (!$task->{saved}) {
 380                 $archive .= sprintf("$task->{upid} %08X $task->{status}\n", $task->{endtime});
 381                 $save = 1;
 382                 push @arlist, $task;
 383                 $task->{saved} = 1;
 384             }
 385         }
 386
 387         if ($archive) {
 388             my $size = 0;
 389             my $filename = "/var/log/pve/tasks/index";
 390             eval {
 391                 my $fh = IO::File->new($filename, '>>', 0644) ||
 392                     die "unable to open file '$filename' - $!\n";
 393                 PVE::Tools::safe_print($filename, $fh, $archive);
 394                 $size = -s $fh;
 395                 close($fh) ||
 396                     die "unable to close file '$filename' - $!\n";
 397             };
 398             my $err = $@;
 399             if ($err) {
 400                 syslog('err', $err);
 401                 foreach my $task (@arlist) { # mark as not saved
 402                     $task->{saved} = 0;
 403                 }
 404             }
 405             my $maxsize = 50000; # about 1000 entries
 406             if ($size > $maxsize) {
 407                 rename($filename, "$filename.1");
 408             }
 409         }
 410
 411         # we try to reduce the amount of data
 412         # list all running tasks and task and a few others
 413         # try to limit to 25 tasks
 414         my $ctime = time();
 415         my $max = 25 - scalar(@$tlist);
 416         foreach my $task (@ta) {
 417             last if $max <= 0;
 418             push @$tlist, $task;
 419             $max--;
 420         }
 421
 422         PVE::INotify::write_file('active', $tlist) if $save;
 423
 424         return $tlist;
 425     };
 426
 427     my $res = PVE::Tools::lock_file($lkfn, $timeout, $code);
 428     die $@ if $@;
 429
 430     return $res;
 431 }
 432
 433 my $kill_process_group = sub {
 434     my ($pid, $pstart) = @_;
 435
 436     # send kill to process group (negative pid)
 437     my $kpid = -$pid;
 438
 439     # always send signal to all pgrp members
 440     kill(15, $kpid); # send TERM signal
 441
 442     # give max 5 seconds to shut down
 443     for (my $i = 0; $i < 5; $i++) {
 444         return if !PVE::ProcFSTools::check_process_running($pid, $pstart);
 445         sleep (1);
 446     }
 447
 448     # to be sure
 449     kill(9, $kpid);
 450 };
 451
 452 sub check_worker {
 453     my ($upid, $killit) = @_;
 454
 455     my $task = PVE::Tools::upid_decode($upid);
 456
 457     my $running = PVE::ProcFSTools::check_process_running($task->{pid}, $task->{pstart});
 458
 459     return 0 if !$running;
 460
 461     if ($killit) {
 462         &$kill_process_group($task->{pid});
 463         return 0;
 464     }
 465
 466     return 1;
 467 }
 468
 469 # start long running workers
 470 # STDIN is redirected to /dev/null
 471 # STDOUT,STDERR are redirected to the filename returned by upid_decode
 472 # NOTE: we simulate running in foreground if ($self->{type} eq 'cli')
 473 sub fork_worker {
 474     my ($self, $dtype, $id, $user, $function) = @_;
 475
 476     $dtype = 'unknown' if !defined ($dtype);
 477     $id = '' if !defined ($id);
 478
 479     $user = 'root@pve' if !defined ($user);
 480
 481     my $sync = $self->{type} eq 'cli' ? 1 : 0;
 482
 483     local $SIG{INT} =
 484         local $SIG{QUIT} =
 485         local $SIG{PIPE} =
 486         local $SIG{TERM} = 'IGNORE';
 487
 488     my $starttime = time ();
 489
 490     my @psync = POSIX::pipe();
 491     my @csync = POSIX::pipe();
 492
 493     my $node = $self->{nodename};
 494
 495     my $cpid = fork();
 496     die "unable to fork worker - $!" if !defined($cpid);
 497
 498     my $workerpuid = $cpid ? $cpid : $$;
 499
 500     my $pstart = PVE::ProcFSTools::read_proc_starttime($workerpuid) ||
 501         die "unable to read process start time";
 502
 503     my $upid = PVE::Tools::upid_encode ({
 504         node => $node, pid => $workerpuid, pstart => $pstart,
 505         starttime => $starttime, type => $dtype, id => $id, user => $user });
 506
 507     my $outfh;
 508
 509     if (!$cpid) { # child
 510
 511         $0 = "task $upid";
 512
 513         $SIG{INT} = $SIG{QUIT} = $SIG{TERM} = sub { die "received interrupt\n"; };
 514
 515         $SIG{CHLD} = $SIG{PIPE} = 'DEFAULT';
 516
 517         # set sess/process group - we want to be able to kill the
 518         # whole process group
 519         POSIX::setsid();
 520
 521         POSIX::close ($psync[0]);
 522         POSIX::close ($csync[1]);
 523
 524         $outfh = $sync ? $psync[1] : undef;
 525
 526         eval {
 527             PVE::INotify::inotify_close();
 528
 529             if (my $atfork = $self->{atfork}) {
 530                 &$atfork();
 531             }
 532
 533             # same algorythm as used inside SA
 534             # STDIN = /dev/null
 535             my $fd = fileno (STDIN);
 536
 537             if (!$sync) {
 538                 close STDIN;
 539                 POSIX::close(0) if $fd != 0;
 540
 541                 die "unable to redirect STDIN - $!"
 542                     if !open(STDIN, "</dev/null");
 543
 544                 $outfh = PVE::Tools::upid_open($upid);
 545             }
 546
 547
 548             # redirect STDOUT
 549             $fd = fileno(STDOUT);
 550             close STDOUT;
 551             POSIX::close (1) if $fd != 1;
 552
 553             die "unable to redirect STDOUT - $!"
 554                 if !open(STDOUT, ">&", $outfh);
 555
 556             STDOUT->autoflush (1);
 557
 558             #  redirect STDERR to STDOUT
 559             $fd = fileno (STDERR);
 560             close STDERR;
 561             POSIX::close(2) if $fd != 2;
 562
 563             die "unable to redirect STDERR - $!"
 564                 if !open(STDERR, ">&1");
 565
 566             STDERR->autoflush(1);
 567         };
 568         if (my $err = $@) {
 569             my $msg =  "ERROR: $err";
 570             POSIX::write($psync[1], $msg, length ($msg));
 571             POSIX::close($psync[1]);
 572             POSIX::_exit(1);
 573             kill(-9, $$);
 574         }
 575
 576         # sync with parent (signal that we are ready)
 577         if ($sync) {
 578             print "$upid\n";
 579         } else {
 580             POSIX::write($psync[1], $upid, length ($upid));
 581             POSIX::close($psync[1]);
 582         }
 583
 584         my $readbuf = '';
 585         # sync with parent (wait until parent is ready)
 586         POSIX::read($csync[0], $readbuf, 4096);
 587         die "parent setup error\n" if $readbuf ne 'OK';
 588
 589         eval { &$function($upid); };
 590         my $err = $@;
 591         if ($err) {
 592             chomp $err;
 593             $err =~ s/\n/ /mg;
 594             syslog('err', $err);
 595             print STDERR "TASK ERROR: $err\n";
 596             POSIX::_exit(-1);
 597         } else {
 598             print STDERR "TASK OK\n";
 599             POSIX::_exit(0);
 600         }
 601         kill(-9, $$);
 602     }
 603
 604     # parent
 605
 606     POSIX::close ($psync[1]);
 607     POSIX::close ($csync[0]);
 608
 609     my $readbuf = '';
 610     # sync with child (wait until child starts)
 611     POSIX::read($psync[0], $readbuf, 4096);
 612
 613     if (!$sync) {
 614         POSIX::close($psync[0]);
 615         &$register_worker($cpid, $user, $upid);
 616     } else {
 617         chomp $readbuf;
 618     }
 619
 620     eval {
 621         die "got no worker upid - start worker failed\n" if !$readbuf;
 622
 623         if ($readbuf =~ m/^ERROR:\s*(.+)$/m) {
 624             die "starting worker failed: $1\n";
 625         }
 626
 627         if ($readbuf ne $upid) {
 628             die "got strange worker upid ('$readbuf' != '$upid') - start worker failed\n";
 629         }
 630
 631         if ($sync) {
 632             $outfh = PVE::Tools::upid_open($upid);
 633         }
 634     };
 635     my $err = $@;
 636
 637     if (!$err) {
 638         my $msg = 'OK';
 639         POSIX::write($csync[1], $msg, length ($msg));
 640         POSIX::close($csync[1]);
 641
 642     } else {
 643         POSIX::close($csync[1]);
 644         kill(-9, $cpid); # make sure it gets killed
 645         die $err;
 646     }
 647
 648     PVE::Cluster::log_msg('info', $user, "starting task $upid");
 649
 650     my $tlist = active_workers($upid, $sync);
 651     PVE::Cluster::broadcast_tasklist($tlist);
 652
 653     my $res = 0;
 654
 655     if ($sync) {
 656         my $count;
 657         my $outbuf = '';
 658         my $int_count = 0;
 659         eval {
 660             local $SIG{INT} = local $SIG{QUIT} = local $SIG{TERM} = sub {
 661                 # always send signal to all pgrp members
 662                 my $kpid = -$cpid;
 663                 if ($int_count < 3) {
 664                     kill(15, $kpid); # send TERM signal
 665                 } else {
 666                     kill(9, $kpid); # send KILL signal
 667                 }
 668                 $int_count++;
 669             };
 670             local $SIG{PIPE} = sub { die "broken pipe\n"; };
 671
 672             my $select = new IO::Select;
 673             my $fh = IO::Handle->new_from_fd($psync[0], 'r');
 674             $select->add($fh);
 675
 676             while ($select->count) {
 677                 my @handles = $select->can_read(1);
 678                 if (scalar(@handles)) {
 679                     my $count = sysread ($handles[0], $readbuf, 4096);
 680                     if (!defined ($count)) {
 681                         my $err = $!;
 682                         die "sync pipe read error: $err\n";
 683                     }
 684                     last if $count == 0; # eof
 685
 686                     $outbuf .= $readbuf;
 687                     while ($outbuf =~ s/^(([^\010\r\n]*)(\r|\n|(\010)+|\r\n))//s) {
 688                         my $line = $1;
 689                         my $data = $2;
 690                         if ($data =~ m/^TASK OK$/) {
 691                             # skip
 692                         } elsif ($data =~ m/^TASK ERROR: (.+)$/) {
 693                             print STDERR "$1\n";
 694                         } else {
 695                             print $line;
 696                         }
 697                         if ($outfh) {
 698                             print $outfh $line;
 699                             $outfh->flush();
 700                         }
 701                     }
 702                 } else {
 703                     # some commands daemonize without closing stdout
 704                     last if !PVE::ProcFSTools::check_process_running($cpid);
 705                 }
 706             }
 707         };
 708         my $err = $@;
 709
 710         POSIX::close($psync[0]);
 711
 712         if ($outbuf) { # just to be sure
 713             print $outbuf;
 714             if ($outfh) {
 715                 print $outfh $outbuf;
 716             }
 717         }
 718
 719         if ($err) {
 720             $err =~ s/\n/ /mg;
 721             print STDERR "$err\n";
 722             if ($outfh) {
 723                 print $outfh "TASK ERROR: $err\n";
 724             }
 725         }
 726
 727         &$kill_process_group($cpid, $pstart); # make sure it gets killed
 728
 729         close($outfh);
 730
 731         waitpid($cpid, 0);
 732         $res = $?;
 733         &$log_task_result($upid, $user, $res);
 734     }
 735
 736     return wantarray ? ($upid, $res) : $upid;
 737 }
 738
 739 1;