src/PMG/Cluster.pm

   1 package PMG::Cluster;
   2
   3 use strict;
   4 use warnings;
   5 use Data::Dumper;
   6 use Socket;
   7 use File::Path;
   8 use Time::HiRes qw (gettimeofday tv_interval);
   9
  10 use PVE::SafeSyslog;
  11 use PVE::Tools;
  12 use PVE::INotify;
  13 use PVE::APIClient::LWP;
  14 use PVE::Network;
  15
  16 use PMG::Utils;
  17 use PMG::Config;
  18 use PMG::ClusterConfig;
  19 use PMG::RuleDB;
  20 use PMG::RuleCache;
  21 use PMG::MailQueue;
  22 use PMG::Fetchmail;
  23 use PMG::Ticket;
  24
  25 sub remote_node_ip {
  26     my ($nodename, $noerr) = @_;
  27
  28     my $cinfo = PMG::ClusterConfig->new();
  29
  30     foreach my $entry (values %{$cinfo->{ids}}) {
  31         if ($entry->{name} eq $nodename) {
  32             my $ip = $entry->{ip};
  33             return $ip if !wantarray;
  34             my $family = PVE::Tools::get_host_address_family($ip);
  35             return ($ip, $family);
  36         }
  37     }
  38
  39     # fallback: try to get IP by other means
  40     if ($nodename eq 'localhost' || $nodename eq PVE::INotify::nodename()) {
  41         return PVE::Network::get_local_ip();
  42     } else {
  43         return PVE::Network::get_ip_from_hostname($nodename, $noerr);
  44     }
  45 }
  46
  47 sub get_master_node {
  48     my ($cinfo) = @_;
  49
  50     $cinfo = PMG::ClusterConfig->new() if !$cinfo;
  51
  52     return $cinfo->{master}->{name} if defined($cinfo->{master});
  53
  54     return 'localhost';
  55 }
  56
  57 sub read_local_ssl_cert_fingerprint {
  58     my $cert_path = "/etc/pmg/pmg-api.pem";
  59
  60     my $cert;
  61     eval {
  62         my $bio = Net::SSLeay::BIO_new_file($cert_path, 'r');
  63         $cert = Net::SSLeay::PEM_read_bio_X509($bio);
  64         Net::SSLeay::BIO_free($bio);
  65     };
  66     if (my $err = $@) {
  67         die "unable to read certificate '$cert_path' - $err\n";
  68     }
  69
  70     if (!defined($cert)) {
  71         die "unable to read certificate '$cert_path' - got empty value\n";
  72     }
  73
  74     my $fp;
  75     eval {
  76         $fp = Net::SSLeay::X509_get_fingerprint($cert, 'sha256');
  77     };
  78     if (my $err = $@) {
  79         die "unable to get fingerprint for '$cert_path' - $err\n";
  80     }
  81
  82     if (!defined($fp) || $fp eq '') {
  83         die "unable to get fingerprint for '$cert_path' - got empty value\n";
  84     }
  85
  86     return $fp;
  87 }
  88
  89 my $hostrsapubkey_fn = '/etc/ssh/ssh_host_rsa_key.pub';
  90 my $rootrsakey_fn = '/root/.ssh/id_rsa';
  91 my $rootrsapubkey_fn = '/root/.ssh/id_rsa.pub';
  92
  93 sub read_local_cluster_info {
  94
  95     my $res = {};
  96
  97     my $hostrsapubkey = PVE::Tools::file_read_firstline($hostrsapubkey_fn);
  98     $hostrsapubkey =~ s/^.*ssh-rsa\s+//i;
  99     $hostrsapubkey =~ s/\s+root\@\S+\s*$//i;
 100
 101     my $sshpubkeypattern = PMG::ClusterConfig::Node::valid_ssh_pubkey_regex();
 102     die "unable to parse ${hostrsapubkey_fn}\n"
 103         if $hostrsapubkey !~ m/$sshpubkeypattern/;
 104
 105     my $nodename = PVE::INotify::nodename();
 106
 107     $res->{name} = $nodename;
 108
 109     $res->{ip} = PVE::Network::get_local_ip();
 110
 111     $res->{hostrsapubkey} = $hostrsapubkey;
 112
 113     if (! -f $rootrsapubkey_fn) {
 114         unlink $rootrsakey_fn;
 115         my $cmd = ['ssh-keygen', '-t', 'rsa', '-N', '', '-b', '2048',
 116                    '-f', $rootrsakey_fn];
 117         PMG::Utils::run_silent_cmd($cmd);
 118     }
 119
 120     my $rootrsapubkey = PVE::Tools::file_read_firstline($rootrsapubkey_fn);
 121     $rootrsapubkey =~ s/^.*ssh-rsa\s+//i;
 122     $rootrsapubkey =~ s/\s+root\@\S+\s*$//i;
 123
 124     die "unable to parse ${rootrsapubkey_fn}\n"
 125         if $rootrsapubkey !~ m/$sshpubkeypattern/;
 126
 127     $res->{rootrsapubkey} = $rootrsapubkey;
 128
 129     $res->{fingerprint} = read_local_ssl_cert_fingerprint();
 130
 131     return $res;
 132 }
 133
 134 # X509 Certificate cache helper
 135
 136 my $cert_cache_nodes = {};
 137 my $cert_cache_timestamp = time();
 138 my $cert_cache_fingerprints = {};
 139
 140 sub update_cert_cache {
 141
 142     $cert_cache_timestamp = time();
 143
 144     $cert_cache_fingerprints = {};
 145     $cert_cache_nodes = {};
 146
 147     my $cinfo = PMG::ClusterConfig->new();
 148
 149     foreach my $entry (values %{$cinfo->{ids}}) {
 150         my $node = $entry->{name};
 151         my $fp = $entry->{fingerprint};
 152         if ($node && $fp) {
 153             $cert_cache_fingerprints->{$fp} = 1;
 154             $cert_cache_nodes->{$node} = $fp;
 155         }
 156     }
 157 }
 158
 159 # load and cache cert fingerprint once
 160 sub initialize_cert_cache {
 161     my ($node) = @_;
 162
 163     update_cert_cache()
 164         if defined($node) && !defined($cert_cache_nodes->{$node});
 165 }
 166
 167 sub check_cert_fingerprint {
 168     my ($cert) = @_;
 169
 170     # clear cache every 30 minutes at least
 171     update_cert_cache() if time() - $cert_cache_timestamp >= 60*30;
 172
 173     # get fingerprint of server certificate
 174     my $fp;
 175     eval {
 176         $fp = Net::SSLeay::X509_get_fingerprint($cert, 'sha256');
 177     };
 178     return 0 if $@ || !defined($fp) || $fp eq ''; # error
 179
 180     my $check = sub {
 181         for my $expected (keys %$cert_cache_fingerprints) {
 182             return 1 if $fp eq $expected;
 183         }
 184         return 0;
 185     };
 186
 187     return 1 if $check->();
 188
 189     # clear cache and retry at most once every minute
 190     if (time() - $cert_cache_timestamp >= 60) {
 191         syslog ('info', "Could not verify remote node certificate '$fp' with list of pinned certificates, refreshing cache");
 192         update_cert_cache();
 193         return $check->();
 194     }
 195
 196     return 0;
 197 }
 198
 199 my $sshglobalknownhosts = "/etc/ssh/ssh_known_hosts2";
 200 my $rootsshauthkeys = "/root/.ssh/authorized_keys";
 201 my $ssh_rsa_id = "/root/.ssh/id_rsa.pub";
 202
 203 sub update_ssh_keys {
 204     my ($cinfo) = @_;
 205
 206     my $old = '';
 207     my $data = '';
 208
 209     foreach my $node (values %{$cinfo->{ids}}) {
 210         $data .= "$node->{ip} ssh-rsa $node->{hostrsapubkey}\n";
 211         $data .= "$node->{name} ssh-rsa $node->{hostrsapubkey}\n";
 212     }
 213
 214     $old = PVE::Tools::file_get_contents($sshglobalknownhosts, 1024*1024)
 215         if -f $sshglobalknownhosts;
 216
 217     PVE::Tools::file_set_contents($sshglobalknownhosts, $data)
 218         if $old ne $data;
 219
 220     $data = '';
 221     $old = '';
 222
 223     # always add ourself
 224     if (-f $ssh_rsa_id) {
 225         my $pub = PVE::Tools::file_get_contents($ssh_rsa_id);
 226         chomp($pub);
 227         $data .= "$pub\n";
 228     }
 229
 230     foreach my $node (values %{$cinfo->{ids}}) {
 231         $data .= "ssh-rsa $node->{rootrsapubkey} root\@$node->{name}\n";
 232     }
 233
 234     if (-f $rootsshauthkeys) {
 235         my $mykey = PVE::Tools::file_get_contents($rootsshauthkeys, 128*1024);
 236         chomp($mykey);
 237         $data .= "$mykey\n";
 238     }
 239
 240     my $newdata = "";
 241     my $vhash = {};
 242     my @lines = split(/\n/, $data);
 243     foreach my $line (@lines) {
 244         if ($line !~ /^#/ && $line =~ m/(^|\s)ssh-(rsa|dsa)\s+(\S+)\s+\S+$/) {
 245             next if $vhash->{$3}++;
 246         }
 247         $newdata .= "$line\n";
 248     }
 249
 250     $old = PVE::Tools::file_get_contents($rootsshauthkeys, 1024*1024)
 251         if -f $rootsshauthkeys;
 252
 253     PVE::Tools::file_set_contents($rootsshauthkeys, $newdata, 0600)
 254         if $old ne $newdata;
 255 }
 256
 257 my $cfgdir = '/etc/pmg';
 258 my $syncdir = "$cfgdir/master";
 259
 260 my $cond_commit_synced_file = sub {
 261     my ($filename, $dstfn) = @_;
 262
 263     $dstfn = "$cfgdir/$filename" if !defined($dstfn);
 264     my $srcfn = "$syncdir/$filename";
 265
 266     if (! -f $srcfn) {
 267         unlink $dstfn;
 268         return;
 269     }
 270
 271     my $new = PVE::Tools::file_get_contents($srcfn, 1024*1024);
 272
 273     if (-f $dstfn) {
 274         my $old = PVE::Tools::file_get_contents($dstfn, 1024*1024);
 275         return 0 if $new eq $old;
 276     }
 277
 278     # set mtime (touch) to avoid time drift problems
 279     utime(undef, undef, $srcfn);
 280
 281     rename($srcfn, $dstfn) ||
 282         die "cond_rename_file '$filename' failed - $!\n";
 283
 284     print STDERR "updated $dstfn\n";
 285
 286     return 1;
 287 };
 288
 289 my $ssh_command = sub {
 290     my ($host_key_alias, @args) = @_;
 291
 292     my $cmd = ['ssh', '-l', 'root', '-o', 'BatchMode=yes'];
 293     push @$cmd, '-o', "HostKeyAlias=${host_key_alias}" if $host_key_alias;
 294     push @$cmd, @args if @args;
 295     return $cmd;
 296 };
 297
 298 sub get_remote_cert_fingerprint {
 299     my ($ni) = @_;
 300
 301     my $ssh_cmd = $ssh_command->(
 302         $ni->{name},
 303         $ni->{ip},
 304         'openssl x509 -noout -fingerprint -sha256 -in /etc/pmg/pmg-api.pem'
 305     );
 306     my $fp;
 307     eval {
 308         PVE::Tools::run_command($ssh_cmd, outfunc => sub {
 309             my ($line) = @_;
 310             if ($line =~ m/SHA256 Fingerprint=((?:[a-f0-9]{2}:){31}[a-f0-9]{2})/i) {
 311                 $fp = $1;
 312             }
 313         });
 314         die "parsing failed\n" if !$fp;
 315     };
 316     die "unable to get remote node fingerprint from '$ni->{name}': $@\n" if $@;
 317
 318     return $fp;
 319 }
 320
 321 sub trigger_update_fingerprints {
 322     my ($cinfo) = @_;
 323
 324     my $master = $cinfo->{master} || die "unable to lookup master node\n";
 325     my $cached_fp = { $master->{fingerprint} => 1 };
 326
 327     # if running on master the current fingerprint for the API-connection is needed
 328     # in addition (to prevent races with restarting pmgproxy
 329     if ($cinfo->{local}->{type} eq 'master') {
 330         my $new_fp = PMG::Cluster::read_local_ssl_cert_fingerprint();
 331         $cached_fp->{$new_fp} = 1;
 332     }
 333
 334     my $ticket = PMG::Ticket::assemble_ticket('root@pam');
 335     my $csrftoken = PMG::Ticket::assemble_csrf_prevention_token('root@pam');
 336     my $conn = PVE::APIClient::LWP->new(
 337         ticket => $ticket,
 338         csrftoken => $csrftoken,
 339         cookie_name => 'PMGAuthCookie',
 340         host => $master->{ip},
 341         cached_fingerprints => $cached_fp,
 342         );
 343
 344     $conn->post("/config/cluster/update-fingerprints", {});
 345     return undef;
 346 }
 347
 348 my $rsync_command = sub {
 349     my ($host_key_alias, @args) = @_;
 350
 351     my $ssh_cmd = join(' ', @{$ssh_command->($host_key_alias)});
 352
 353     my $cmd = ['rsync', "--rsh=$ssh_cmd",  '-q', @args];
 354
 355     return $cmd;
 356 };
 357
 358 sub sync_quarantine_files {
 359     my ($host_ip, $host_name, $flistname, $rcid) = @_;
 360
 361     my $spooldir = $PMG::MailQueue::spooldir;
 362
 363     mkdir "$spooldir/cluster/";
 364     my $syncdir = "$spooldir/cluster/$rcid";
 365     mkdir $syncdir;
 366
 367     my $cmd = $rsync_command->(
 368         $host_name, '--timeout', '10', "[${host_ip}]:$spooldir", $spooldir,
 369         '--files-from', $flistname);
 370
 371     PVE::Tools::run_command($cmd);
 372 }
 373
 374 sub sync_spooldir {
 375     my ($host_ip, $host_name, $rcid) = @_;
 376
 377     my $spooldir = $PMG::MailQueue::spooldir;
 378
 379     mkdir "$spooldir/cluster/";
 380     my $syncdir = "$spooldir/cluster/$rcid";
 381     mkdir $syncdir;
 382
 383     my $cmd = $rsync_command->(
 384         $host_name, '-aq', '--timeout', '10', "[${host_ip}]:$syncdir/", $syncdir);
 385
 386     foreach my $incl (('spam/', 'spam/*', 'spam/*/*', 'virus/', 'virus/*', 'virus/*/*')) {
 387         push @$cmd, '--include', $incl;
 388     }
 389
 390     push @$cmd, '--exclude', '*';
 391
 392     PVE::Tools::run_command($cmd);
 393 }
 394
 395 sub sync_master_quar {
 396     my ($host_ip, $host_name) = @_;
 397
 398     my $spooldir = $PMG::MailQueue::spooldir;
 399
 400     my $syncdir = "$spooldir/cluster/";
 401     mkdir $syncdir;
 402
 403     my $cmd = $rsync_command->(
 404         $host_name, '-aq', '--timeout', '10', "[${host_ip}]:$syncdir", $syncdir);
 405
 406     PVE::Tools::run_command($cmd);
 407 }
 408
 409 sub sync_config_from_master {
 410     my ($master_name, $master_ip, $noreload) = @_;
 411
 412     mkdir $syncdir;
 413     File::Path::remove_tree($syncdir, {keep_root => 1});
 414
 415     my $sa_conf_dir = "/etc/mail/spamassassin";
 416     my $sa_custom_cf = "custom.cf";
 417     my $sa_rules_cf = "pmg-scores.cf";
 418
 419     my $cmd = $rsync_command->(
 420         $master_name, '-aq',
 421         "[${master_ip}]:$cfgdir/*",
 422         "[${master_ip}]:${sa_conf_dir}/${sa_custom_cf}",
 423         "[${master_ip}]:${sa_conf_dir}/${sa_rules_cf}",
 424         "$syncdir/",
 425         '--exclude', 'master/',
 426         '--exclude', '*~',
 427         '--exclude', '*.db',
 428         '--exclude', 'pmg-api.pem',
 429         '--exclude', 'pmg-tls.pem',
 430         );
 431
 432     my $errmsg = "syncing master configuration from '${master_ip}' failed";
 433     PVE::Tools::run_command($cmd, errmsg => $errmsg);
 434
 435     # verify that the remote host is cluster master
 436     open (my $fh, '<', "$syncdir/cluster.conf") ||
 437         die "unable to open synced cluster.conf - $!\n";
 438
 439     my $cinfo = PMG::ClusterConfig::read_cluster_conf('cluster.conf', $fh);
 440
 441     if (!$cinfo->{master} || ($cinfo->{master}->{ip} ne $master_ip)) {
 442         die "host '$master_ip' is not cluster master\n";
 443     }
 444
 445     my $role = $cinfo->{'local'}->{type} // '-';
 446     die "local node '$cinfo->{local}->{name}' not part of cluster\n"
 447         if $role eq '-';
 448
 449     die "local node '$cinfo->{local}->{name}' is new cluster master\n"
 450         if $role eq 'master';
 451
 452     $cond_commit_synced_file->('cluster.conf');
 453
 454     update_ssh_keys($cinfo); # rewrite ssh keys
 455
 456     PMG::Fetchmail::update_fetchmail_default(0); # disable on slave
 457
 458     my $files = [
 459         'pmg-authkey.key',
 460         'pmg-authkey.pub',
 461         'pmg-csrf.key',
 462         'ldap.conf',
 463         'user.conf',
 464         'tfa.json',
 465         'domains',
 466         'mynetworks',
 467         'transport',
 468         'tls_policy',
 469         'tls_inbound_domains',
 470         'fetchmailrc',
 471         ];
 472
 473     foreach my $filename (@$files) {
 474         $cond_commit_synced_file->($filename);
 475     }
 476
 477     my $dirs = [
 478         'templates',
 479         'dkim',
 480         'pbs',
 481         'acme',
 482     ];
 483
 484     foreach my $dir (@$dirs) {
 485         my $srcdir = "$syncdir/$dir";
 486
 487         if ( -d $srcdir ) {
 488             my $cmd = ['rsync', '-aq', '--delete-after', "$srcdir/", "$cfgdir/$dir"];
 489             PVE::Tools::run_command($cmd);
 490         }
 491
 492     }
 493
 494     my $force_restart = {};
 495
 496     for my $file (($sa_custom_cf, $sa_rules_cf)) {
 497         if ($cond_commit_synced_file->($file, "${sa_conf_dir}/${file}")) {
 498             $force_restart->{'pmg-smtp-filter'} = 1;
 499         }
 500     }
 501
 502     $cond_commit_synced_file->('pmg.conf');
 503
 504     return $force_restart;
 505 }
 506
 507 sub sync_ruledb_from_master {
 508     my ($ldb, $rdb, $ni, $ticket) = @_;
 509
 510     my $ruledb = PMG::RuleDB->new($ldb);
 511     my $rulecache = PMG::RuleCache->new($ruledb);
 512
 513     my $conn = PVE::APIClient::LWP->new(
 514         ticket => $ticket,
 515         cookie_name => 'PMGAuthCookie',
 516         host => $ni->{ip},
 517         cached_fingerprints => {
 518             $ni->{fingerprint} => 1,
 519         });
 520
 521     my $digest = $conn->get("/config/ruledb/digest", {});
 522
 523     return if $digest eq $rulecache->{digest}; # no changes
 524
 525     syslog('info', "detected rule database changes - starting sync from '$ni->{ip}'");
 526
 527     eval {
 528         $ldb->begin_work;
 529
 530         $ldb->do("DELETE FROM Rule");
 531         $ldb->do("DELETE FROM RuleGroup");
 532         $ldb->do("DELETE FROM ObjectGroup");
 533         $ldb->do("DELETE FROM Object");
 534         $ldb->do("DELETE FROM Attribut");
 535         $ldb->do("DELETE FROM Objectgroup_Attributes");
 536         $ldb->do("DELETE FROM Rule_Attributes");
 537
 538         eval {
 539             $rdb->begin_work;
 540
 541             # read a consistent snapshot
 542             $rdb->do("SET TRANSACTION ISOLATION LEVEL SERIALIZABLE");
 543
 544             PMG::DBTools::copy_table($ldb, $rdb, "Rule");
 545             PMG::DBTools::copy_table($ldb, $rdb, "RuleGroup");
 546             PMG::DBTools::copy_table($ldb, $rdb, "ObjectGroup");
 547             PMG::DBTools::copy_table($ldb, $rdb, "Object", 'value');
 548             PMG::DBTools::copy_table($ldb, $rdb, "Attribut", 'value');
 549             PMG::DBTools::copy_table($ldb, $rdb, "Rule_Attributes");
 550             PMG::DBTools::copy_table($ldb, $rdb, "Objectgroup_Attributes");
 551         };
 552
 553         $rdb->rollback; # end transaction
 554
 555         die $@ if $@;
 556
 557         # update sequences
 558
 559         $ldb->do("SELECT setval('rule_id_seq', max(id)+1) FROM Rule");
 560         $ldb->do("SELECT setval('object_id_seq', max(id)+1) FROM Object");
 561         $ldb->do("SELECT setval('objectgroup_id_seq', max(id)+1) FROM ObjectGroup");
 562
 563         $ldb->commit;
 564     };
 565     if (my $err = $@) {
 566         $ldb->rollback;
 567         die $err;
 568     }
 569
 570     PMG::DBTools::reload_ruledb();
 571
 572     syslog('info', "finished rule database sync from host '$ni->{ip}'");
 573 }
 574
 575 sub sync_quarantine_db {
 576     my ($ldb, $rdb, $ni, $rsynctime_ref) = @_;
 577
 578     my $rcid = $ni->{cid};
 579
 580     my $maxmails = 100000;
 581
 582     my $mscount = 0;
 583
 584     my $ctime = PMG::DBTools::get_remote_time($rdb);
 585
 586     my $maxcount = 1000;
 587
 588     my $count;
 589
 590     PMG::DBTools::create_clusterinfo_default($ldb, $rcid, 'lastid_CMailStore', -1, undef);
 591
 592     do { # get new values
 593
 594         $count = 0;
 595
 596         my $flistname = "/tmp/quarantinefilelist.$$";
 597
 598         eval {
 599             $ldb->begin_work;
 600
 601             open(my $flistfh, '>', $flistname) ||
 602                 die "unable to open file '$flistname' - $!\n";
 603
 604             my $lastid = PMG::DBTools::read_int_clusterinfo($ldb, $rcid, 'lastid_CMailStore');
 605
 606             # sync CMailStore
 607
 608             my $sth = $rdb->prepare(
 609                 "SELECT * from CMailstore WHERE cid = ? AND rid > ? " .
 610                 "ORDER BY cid,rid LIMIT ?");
 611             $sth->execute($rcid, $lastid, $maxcount);
 612
 613             my $maxid;
 614             my $callback = sub {
 615                 my $ref = shift;
 616                 $maxid = $ref->{rid};
 617                 my $filename = $ref->{file};
 618                  # skip files generated before cluster was created
 619                 return if $filename !~ m!^cluster/!;
 620                 print $flistfh "$filename\n";
 621             };
 622
 623             my $attrs = [qw(cid rid time qtype bytes spamlevel info sender header file)];
 624             $count += PMG::DBTools::copy_selected_data($ldb, $sth, 'CMailStore', $attrs, $callback);
 625
 626             close($flistfh);
 627
 628             my $starttime = [ gettimeofday() ];
 629             sync_quarantine_files($ni->{ip}, $ni->{name}, $flistname, $rcid);
 630             $$rsynctime_ref += tv_interval($starttime);
 631
 632             if ($maxid) {
 633                 # sync CMSReceivers
 634
 635                 $sth = $rdb->prepare(
 636                     "SELECT * from CMSReceivers WHERE " .
 637                     "CMailStore_CID = ? AND CMailStore_RID > ?  " .
 638                     "AND CMailStore_RID <= ?");
 639                 $sth->execute($rcid, $lastid, $maxid);
 640
 641                 $attrs = [qw(cmailstore_cid cmailstore_rid pmail receiver ticketid status mtime)];
 642                 PMG::DBTools::copy_selected_data($ldb, $sth, 'CMSReceivers', $attrs);
 643
 644                 PMG::DBTools::write_maxint_clusterinfo($ldb, $rcid, 'lastid_CMailStore', $maxid);
 645             }
 646
 647             $ldb->commit;
 648         };
 649         my $err = $@;
 650
 651         unlink $flistname;
 652
 653         if ($err) {
 654             $ldb->rollback;
 655             die $err;
 656         }
 657
 658         $mscount += $count;
 659
 660     } while (($count >= $maxcount) && ($mscount < $maxmails));
 661
 662     PMG::DBTools::create_clusterinfo_default($ldb, $rcid, 'lastmt_CMSReceivers', 0, undef);
 663
 664     eval { # synchronize status updates
 665         $ldb->begin_work;
 666
 667         my $lastmt = PMG::DBTools::read_int_clusterinfo($ldb, $rcid, 'lastmt_CMSReceivers');
 668
 669         my $sth = $rdb->prepare ("SELECT * from CMSReceivers WHERE mtime >= ? AND status != 'N'");
 670         $sth->execute($lastmt);
 671
 672         my $update_sth = $ldb->prepare(
 673             "UPDATE CMSReceivers SET status = ? WHERE " .
 674             "CMailstore_CID = ? AND CMailstore_RID = ? AND TicketID = ?");
 675         while (my $ref = $sth->fetchrow_hashref()) {
 676             $update_sth->execute($ref->{status}, $ref->{cmailstore_cid},
 677                                  $ref->{cmailstore_rid}, $ref->{ticketid});
 678         }
 679
 680         PMG::DBTools::write_maxint_clusterinfo($ldb, $rcid, 'lastmt_CMSReceivers', $ctime);
 681
 682         $ldb->commit;
 683     };
 684     if (my $err = $@) {
 685         $ldb->rollback;
 686         die $err;
 687     }
 688
 689     return $mscount;
 690 }
 691
 692 sub sync_statistic_db {
 693     my ($ldb, $rdb, $ni) = @_;
 694
 695     my $rcid = $ni->{cid};
 696
 697     my $maxmails = 100000;
 698
 699     my $mscount = 0;
 700
 701     my $maxcount = 1000;
 702
 703     my $count;
 704
 705     PMG::DBTools::create_clusterinfo_default(
 706         $ldb, $rcid, 'lastid_CStatistic', -1, undef);
 707
 708     do { # get new values
 709
 710         $count = 0;
 711
 712         eval {
 713             $ldb->begin_work;
 714
 715             my $lastid = PMG::DBTools::read_int_clusterinfo(
 716                 $ldb, $rcid, 'lastid_CStatistic');
 717
 718             # sync CStatistic
 719
 720             my $sth = $rdb->prepare(
 721                 "SELECT * from CStatistic " .
 722                 "WHERE cid = ? AND rid > ? " .
 723                 "ORDER BY cid, rid LIMIT ?");
 724             $sth->execute($rcid, $lastid, $maxcount);
 725
 726             my $maxid;
 727             my $callback = sub {
 728                 my $ref = shift;
 729                 $maxid = $ref->{rid};
 730             };
 731
 732             my $attrs = [qw(cid rid time bytes direction spamlevel ptime virusinfo sender)];
 733             $count += PMG::DBTools::copy_selected_data($ldb, $sth, 'CStatistic', $attrs, $callback);
 734
 735             if ($maxid) {
 736                 # sync CReceivers
 737
 738                 $sth = $rdb->prepare(
 739                     "SELECT * from CReceivers WHERE " .
 740                     "CStatistic_CID = ? AND CStatistic_RID > ? AND CStatistic_RID <= ?");
 741                 $sth->execute($rcid, $lastid, $maxid);
 742
 743                 $attrs = [qw(cstatistic_cid cstatistic_rid blocked receiver)];
 744                 PMG::DBTools::copy_selected_data($ldb, $sth, 'CReceivers', $attrs);
 745             }
 746
 747             PMG::DBTools::write_maxint_clusterinfo ($ldb, $rcid, 'lastid_CStatistic', $maxid);
 748
 749             $ldb->commit;
 750         };
 751         if (my $err = $@) {
 752             $ldb->rollback;
 753             die $err;
 754         }
 755
 756         $mscount += $count;
 757
 758     } while (($count >= $maxcount) && ($mscount < $maxmails));
 759
 760     return $mscount;
 761 }
 762
 763 my $sync_generic_mtime_db = sub {
 764     my ($ldb, $rdb, $ni, $table, $selectfunc, $mergefunc) = @_;
 765
 766     my $ctime = PMG::DBTools::get_remote_time($rdb);
 767
 768     PMG::DBTools::create_clusterinfo_default($ldb, $ni->{cid}, "lastmt_$table", 0, undef);
 769
 770     my $lastmt = PMG::DBTools::read_int_clusterinfo($ldb, $ni->{cid}, "lastmt_$table");
 771
 772     my $sql_cmd = $selectfunc->($ctime, $lastmt);
 773
 774     my $sth = $rdb->prepare($sql_cmd);
 775
 776     $sth->execute();
 777
 778     my $updates = 0;
 779
 780     eval {
 781         # use transaction to speedup things
 782         my $max = 1000; # UPDATE MAX ENTRIES AT ONCE
 783         my $count = 0;
 784         while (my $ref = $sth->fetchrow_hashref()) {
 785             $ldb->begin_work if !$count;
 786             $mergefunc->($ref);
 787             if (++$count >= $max) {
 788                 $count = 0;
 789                 $ldb->commit;
 790             }
 791             $updates++;
 792         }
 793
 794         $ldb->commit if $count;
 795     };
 796     if (my $err = $@) {
 797         $ldb->rollback;
 798         die $err;
 799     }
 800
 801     PMG::DBTools::write_maxint_clusterinfo($ldb, $ni->{cid}, "lastmt_$table", $ctime);
 802
 803     return $updates;
 804 };
 805
 806 sub sync_localstat_db {
 807     my ($dbh, $rdb, $ni) = @_;
 808
 809     my $rcid = $ni->{cid};
 810
 811     my $selectfunc = sub {
 812         my ($ctime, $lastmt) = @_;
 813         return "SELECT * from LocalStat WHERE mtime >= $lastmt AND cid = $rcid";
 814     };
 815
 816     my $merge_sth = $dbh->prepare(
 817         'INSERT INTO LocalStat (Time, RBLCount, PregreetCount, CID, MTime) ' .
 818         'VALUES (?, ?, ?, ?, ?) ' .
 819         'ON CONFLICT (Time, CID) DO UPDATE SET ' .
 820         'RBLCount = excluded.RBLCount, PregreetCount = excluded.PregreetCount, MTime = excluded.MTime');
 821
 822     my $mergefunc = sub {
 823         my ($ref) = @_;
 824
 825         $merge_sth->execute($ref->{time}, $ref->{rblcount}, $ref->{pregreetcount}, $ref->{cid}, $ref->{mtime});
 826     };
 827
 828     return $sync_generic_mtime_db->($dbh, $rdb, $ni, 'LocalStat', $selectfunc, $mergefunc);
 829 }
 830
 831 sub sync_greylist_db {
 832     my ($dbh, $rdb, $ni) = @_;
 833
 834     my $selectfunc = sub {
 835         my ($ctime, $lastmt) = @_;
 836         return "SELECT * from CGreylist WHERE extime >= $ctime AND " .
 837             "mtime >= $lastmt AND CID != 0";
 838     };
 839
 840     my $merge_sth = $dbh->prepare(PMG::DBTools::cgreylist_merge_sql());
 841     my $mergefunc = sub {
 842         my ($ref) = @_;
 843
 844         my $ipnet = $ref->{ipnet};
 845         $ipnet .= '.0/24' if $ipnet !~ /\/\d+$/;
 846         $merge_sth->execute(
 847             $ipnet, $ref->{sender}, $ref->{receiver},
 848             $ref->{instance}, $ref->{rctime}, $ref->{extime}, $ref->{delay},
 849             $ref->{blocked}, $ref->{passed}, 0, $ref->{cid});
 850     };
 851
 852     return $sync_generic_mtime_db->($dbh, $rdb, $ni, 'CGreylist', $selectfunc, $mergefunc);
 853 }
 854
 855 sub sync_userprefs_db {
 856     my ($dbh, $rdb, $ni) = @_;
 857
 858     my $selectfunc = sub {
 859         my ($ctime, $lastmt) = @_;
 860
 861         return "SELECT * from UserPrefs WHERE mtime >= $lastmt";
 862     };
 863
 864     my $merge_sth = $dbh->prepare(
 865         "INSERT INTO UserPrefs (PMail, Name, Data, MTime) " .
 866         'VALUES (?, ?, ?, ?) ' .
 867         'ON CONFLICT (PMail, Name) DO UPDATE SET ' .
 868         # Note: MTime = 0 ==> this is just a copy from somewhere else, not modified
 869         'MTime = CASE WHEN excluded.MTime >= UserPrefs.MTime THEN 0 ELSE UserPrefs.MTime END, ' .
 870         'Data = CASE WHEN excluded.MTime >= UserPrefs.MTime THEN excluded.Data ELSE UserPrefs.Data END');
 871
 872     my $mergefunc = sub {
 873         my ($ref) = @_;
 874
 875         $merge_sth->execute($ref->{pmail}, $ref->{name}, $ref->{data}, $ref->{mtime});
 876     };
 877
 878     return $sync_generic_mtime_db->($dbh, $rdb, $ni, 'UserPrefs', $selectfunc, $mergefunc);
 879 }
 880
 881 sub sync_domainstat_db {
 882     my ($dbh, $rdb, $ni) = @_;
 883
 884     my $selectfunc = sub {
 885         my ($ctime, $lastmt) = @_;
 886         return "SELECT * from DomainStat WHERE mtime >= $lastmt";
 887     };
 888
 889     my $merge_sth = $dbh->prepare(
 890         'INSERT INTO Domainstat ' .
 891         '(Time,Domain,CountIn,CountOut,BytesIn,BytesOut,VirusIn,VirusOut,SpamIn,SpamOut,' .
 892         'BouncesIn,BouncesOut,PTimeSum,Mtime) ' .
 893         'VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?) ' .
 894         'ON CONFLICT (Time, Domain) DO UPDATE SET ' .
 895         'CountIn = excluded.CountIn, CountOut = excluded.CountOut, ' .
 896         'BytesIn = excluded.BytesIn, BytesOut = excluded.BytesOut, ' .
 897         'VirusIn = excluded.VirusIn, VirusOut = excluded.VirusOut, ' .
 898         'SpamIn = excluded.SpamIn, SpamOut = excluded.SpamOut, ' .
 899         'BouncesIn = excluded.BouncesIn, BouncesOut = excluded.BouncesOut, ' .
 900         'PTimeSum = excluded.PTimeSum, MTime = excluded.MTime');
 901
 902     my $mergefunc = sub {
 903         my ($ref) = @_;
 904
 905         $merge_sth->execute(
 906             $ref->{time}, $ref->{domain}, $ref->{countin}, $ref->{countout},
 907             $ref->{bytesin}, $ref->{bytesout},
 908             $ref->{virusin}, $ref->{virusout}, $ref->{spamin}, $ref->{spamout},
 909             $ref->{bouncesin}, $ref->{bouncesout}, $ref->{ptimesum}, $ref->{mtime});
 910     };
 911
 912     return $sync_generic_mtime_db->($dbh, $rdb, $ni, 'DomainStat', $selectfunc, $mergefunc);
 913 }
 914
 915 sub sync_dailystat_db {
 916     my ($dbh, $rdb, $ni) = @_;
 917
 918     my $selectfunc = sub {
 919         my ($ctime, $lastmt) = @_;
 920         return "SELECT * from DailyStat WHERE mtime >= $lastmt";
 921     };
 922
 923     my $merge_sth = $dbh->prepare(
 924         'INSERT INTO DailyStat ' .
 925         '(Time,CountIn,CountOut,BytesIn,BytesOut,VirusIn,VirusOut,SpamIn,SpamOut,' .
 926         'BouncesIn,BouncesOut,GreylistCount,SPFCount,RBLCount,PTimeSum,Mtime) ' .
 927         'VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) ' .
 928         'ON CONFLICT (Time) DO UPDATE SET ' .
 929         'CountIn = excluded.CountIn, CountOut = excluded.CountOut, ' .
 930         'BytesIn = excluded.BytesIn, BytesOut = excluded.BytesOut, ' .
 931         'VirusIn = excluded.VirusIn, VirusOut = excluded.VirusOut, ' .
 932         'SpamIn = excluded.SpamIn, SpamOut = excluded.SpamOut, ' .
 933         'BouncesIn = excluded.BouncesIn, BouncesOut = excluded.BouncesOut, ' .
 934         'GreylistCount = excluded.GreylistCount, SPFCount = excluded.SpfCount, ' .
 935         'RBLCount = excluded.RBLCount, ' .
 936         'PTimeSum = excluded.PTimeSum, MTime = excluded.MTime');
 937
 938     my $mergefunc = sub {
 939         my ($ref) = @_;
 940
 941         $merge_sth->execute(
 942             $ref->{time}, $ref->{countin}, $ref->{countout},
 943             $ref->{bytesin}, $ref->{bytesout},
 944             $ref->{virusin}, $ref->{virusout}, $ref->{spamin}, $ref->{spamout},
 945             $ref->{bouncesin}, $ref->{bouncesout}, $ref->{greylistcount},
 946             $ref->{spfcount}, $ref->{rblcount}, $ref->{ptimesum}, $ref->{mtime});
 947     };
 948
 949     return $sync_generic_mtime_db->($dbh, $rdb, $ni, 'DailyStat', $selectfunc, $mergefunc);
 950 }
 951
 952 sub sync_virusinfo_db {
 953     my ($dbh, $rdb, $ni) = @_;
 954
 955     my $selectfunc = sub {
 956         my ($ctime, $lastmt) = @_;
 957         return "SELECT * from VirusInfo WHERE mtime >= $lastmt";
 958     };
 959
 960     my $merge_sth = $dbh->prepare(
 961         'INSERT INTO VirusInfo (Time,Name,Count,MTime) ' .
 962         'VALUES (?,?,?,?) ' .
 963         'ON CONFLICT (Time,Name) DO UPDATE SET ' .
 964         'Count = excluded.Count , MTime = excluded.MTime');
 965
 966     my $mergefunc = sub {
 967         my ($ref) = @_;
 968
 969         $merge_sth->execute($ref->{time}, $ref->{name}, $ref->{count}, $ref->{mtime});
 970     };
 971
 972     return $sync_generic_mtime_db->($dbh, $rdb, $ni, 'VirusInfo', $selectfunc, $mergefunc);
 973 }
 974
 975 sub sync_deleted_nodes_from_master {
 976     my ($ldb, $masterdb, $cinfo, $masterni, $rsynctime_ref) = @_;
 977
 978     my $rsynctime = 0;
 979
 980     my $cid_hash = {}; # fast lookup
 981     foreach my $ni (values %{$cinfo->{ids}}) {
 982         $cid_hash->{$ni->{cid}} = $ni;
 983     }
 984
 985     my $spooldir = $PMG::MailQueue::spooldir;
 986
 987     my $maxcid = $cinfo->{master}->{maxcid} // 0;
 988
 989     for (my $rcid = 1; $rcid <= $maxcid; $rcid++) {
 990         next if $cid_hash->{$rcid};
 991
 992         my $done_marker = "$spooldir/cluster/$rcid/.synced-deleted-node";
 993
 994         next if -f $done_marker; # already synced
 995
 996         syslog('info', "syncing deleted node $rcid from master '$masterni->{ip}'");
 997
 998         my $starttime = [ gettimeofday() ];
 999         sync_spooldir($masterni->{ip}, $masterni->{name}, $rcid);
1000         $$rsynctime_ref += tv_interval($starttime);
1001
1002         my $fake_ni = {
1003             ip => $masterni->{ip},
1004             name => $masterni->{name},
1005             cid => $rcid,
1006         };
1007
1008         sync_quarantine_db($ldb, $masterdb, $fake_ni);
1009
1010         sync_statistic_db ($ldb, $masterdb, $fake_ni);
1011
1012         open(my $fh, ">>",  $done_marker);
1013    }
1014 }
1015
1016
1017 1;