Net-Clacks
view release on metacpan or search on metacpan
lib/Net/Clacks/Server.pm view on Meta::CPAN
package Net::Clacks::Server;
#---AUTOPRAGMASTART---
use v5.36;
use strict;
use diagnostics;
use mro 'c3';
use English qw(-no_match_vars);
use Carp qw[carp croak confess cluck longmess shortmess];
our $VERSION = 36;
use autodie qw( close );
use Array::Contains;
use utf8;
use Encode qw(is_utf8 encode_utf8 decode_utf8);
use Data::Dumper;
use builtin qw[true false is_bool];
no warnings qw(experimental::builtin); ## no critic (TestingAndDebugging::ProhibitNoWarnings)
#---AUTOPRAGMAEND---
use XML::Simple;
use Time::HiRes qw(sleep usleep time);
use Sys::Hostname;
use Errno;
use IO::Socket::IP;
use IO::Select;
use IO::Socket::SSL;
use YAML::Syck;
use MIME::Base64;
use File::Copy;
use Scalar::Util qw(looks_like_number weaken);
# For turning off SSL session cache
use Readonly;
Readonly my $SSL_SESS_CACHE_OFF => 0x0000;
my %overheadflags = (
A => "auth_token", # Authentication token
O => "auth_ok", # Authentication OK
F => "auth_failed", # Authentication FAILED
E => 'error_message', # Server to client error message
C => "close_all_connections",
D => "discard_message",
G => "forward_message",
I => "set_interclacks_mode", # value: true/false, disables 'G' and 'U'
L => "lock_for_sync", # value: true/false, only available in interclacks client mode
M => "informal_message", # informal message, no further operation on it
N => "no_logging",
S => "shutdown_service", # value: positive number (number in seconds before shutdown). If interclacks clients are present, should be high
# enough to flush all buffers to them
T => 'timestamp', # Used before KEYSYNC to compensate for time drift between different systems
U => "return_to_sender",
Z => "no_flags", # Only sent when no other flags are set
);
BEGIN {
{
# We need to add some extra function to IO::Socket::SSL so we can track the client ID
# on both TCP and Unix Domain Sockets
no strict 'refs'; ## no critic (TestingAndDebugging::ProhibitNoStrict)
*{"IO::Socket::SSL::_setClientID"} = sub {
my ($self, $cid) = @_;
${*$self}{'__client_id'} = $cid; ## no critic (References::ProhibitDoubleSigils)
return;
};
*{"IO::Socket::SSL::_getClientID"} = sub {
my ($self) = @_;
return ${*$self}{'__client_id'} || ''; ## no critic (References::ProhibitDoubleSigils)
};
}
}
sub new($class, $isDebugging, $configfile) {
my $self = bless {}, $class;
$self->{isDebugging} = $isDebugging;
$self->{configfile} = $configfile;
$self->{timeoffset} = 0;
if(defined($ENV{CLACKS_SIMULATED_TIME_OFFSET})) {
$self->{timeoffset} = 0 + $ENV{CLACKS_SIMULATED_TIME_OFFSET};
print "****** RUNNING WITH A SIMULATED TIME OFFSET OF ", $self->{timeoffset}, " seconds ******\n";
}
$self->{cache} = {};
return $self;
}
sub init($self) {
# Dummy function for backward compatibility
carp("Deprecated call to init(), you can remove that function from your code");
return;
}
sub run($self) {
if(!defined($self->{initHasRun}) || !$self->{initHasRun}) {
$self->_init();
}
my $nextdebugfile = 0;
while($self->{keepRunning}) {
my $now = time;
if($now > $nextdebugfile) {
$nextdebugfile = $now + 10;
$self->_saveDebugFile();
}
# Check for shutdown time
if($self->{shutdowntime} && $self->{shutdowntime} < $now) {
print STDERR "Shutdown time has arrived!\n";
$self->{keepRunning} = 0;
}
$self->runOnce();
if($self->{workCount}) {
$self->{usleep} = 0;
} elsif($self->{usleep} < $self->{config}->{throttle}->{maxsleep}) {
lib/Net/Clacks/Server.pm view on Meta::CPAN
if($inmsg =~ /^OVERHEAD\ /) {
# Already handled
next;
} elsif($self->_handleMessageDirect($cid, $inmsg)) {
# Fallthrough
} elsif($self->_handleMessageCaching($cid, $inmsg)) {
# Fallthrough
} elsif($self->_handleMessageControl($cid, $inmsg)) {
# Fallthrough
# local managment commands
} else {
print STDERR "ERROR Unknown_command ", $inmsg, "\r\n";
$self->{sendinterclacks} = 0;
$self->{clients}->{$cid}->{outbuffer} .= "OVERHEAD E unknown_command " . $inmsg . "\r\n";
}
# forward interclacks messages
if($self->{sendinterclacks}) {
foreach my $interclackscid (keys %{$self->{clients}}) {
if($cid eq $interclackscid || !$self->{clients}->{$interclackscid}->{interclacks}) {
next;
}
$self->{clients}->{$interclackscid}->{outbuffer} .= $inmsg . "\r\n";
}
}
}
}
# Clean up algorithm, only run every so often
if($self->{nextcachecleanup} < $now) {
$self->_cacheCleanup();
$self->{nextcachecleanup} = $now + $self->{config}->{cachecleaninterval};
}
$self->_outboxToClientBuffer();
$self->_clientOutput();
return $self->{workCount};
}
sub runShutdown($self) {
print "Shutting down...\n";
# Make sure we save the latest version of the persistance file
$self->_savePersistanceFile();
sleep(0.5);
foreach my $cid (keys %{$self->{clients}}) {
print "Removing client $cid\n";
# Try to notify the client (may or may not work);
$self->_evalsyswrite($self->{clients}->{$cid}->{socket}, "\r\nQUIT\r\n");
# Remove from the selector AND explicitly close — relying on Perl GC was
# leaving handles (and thus OS file descriptors) alive whenever IO::Select
# still held a reference, particularly for IO::Socket::SSL handles.
eval { ## no critic (ErrorHandling::RequireCheckingReturnValueOfEval)
$self->{selector}->remove($self->{clients}->{$cid}->{socket});
};
$self->_safeCloseSocket($self->{clients}->{$cid}->{socket});
delete $self->{clients}->{$cid};
}
print "All clients removed\n";
# Close the listening sockets too so the next process can re-bind without
# waiting for the kernel to release the port / unix-socket inode.
foreach my $listener (@{$self->{tcpsockets}}) {
$self->_safeCloseSocket($listener);
}
$self->{tcpsockets} = [];
return;
}
sub _saveDebugFile($self) {
if(defined($self->{config}->{debugfile})) {
if(open(my $ofh, '>', $self->{config}->{debugfile})) {
print $ofh Dumper($self->{clients});
close $ofh;
}
}
return;
}
sub _savePersistanceFile($self) {
if(!$self->{persistance}) {
return;
}
print "Saving persistance file\n";
my $tempfname = $self->{config}->{persistancefile} . '_';
my $backfname = $self->{config}->{persistancefile} . '_bck';
if($self->{savecache} == 1) {
# Normal savecache operation only
copy($self->{config}->{persistancefile}, $backfname);
}
my $persistancedata = chr(0) . 'CLACKSV3' . Dump($self->{cache}) . chr(0) . 'CLACKSV3';
$self->_writeBinFile($tempfname, $persistancedata);
move($tempfname, $self->{config}->{persistancefile});
if($self->{savecache} == 2) {
# Need to make sure we have a valid backup file, since we had a general problem while loading
copy($self->{config}->{persistancefile}, $backfname);
}
return;
}
sub _evalsyswrite($self, $socket, $buffer) {
return false unless(length($buffer));
my $written = 0;
my $ok = 0;
eval { ## no critic (ErrorHandling::RequireCheckingReturnValueOfEval)
$written = syswrite($socket, $buffer);
$ok = 1;
};
if($EVAL_ERROR || !$ok) {
print STDERR "Write error: $EVAL_ERROR\n";
return -1;
}
return $written;
}
sub _getTime($self) {
my $now = time + $self->{timeoffset};
return $now;
}
# Returns a process-monotonic serial number used to make client CIDs unique.
sub _nextClientSerial($self) {
$self->{clientSerial}++;
return $self->{clientSerial};
}
# Best-effort socket close that never throws. SSL sockets get SSL_no_shutdown so
# we don't block trying to send a TLS close_notify to a peer that's already gone.
sub _safeCloseSocket($self, $socket) {
return if(!defined($socket));
eval { ## no critic (ErrorHandling::RequireCheckingReturnValueOfEval)
if(ref($socket) =~ /^IO::Socket::SSL/) {
$socket->close(SSL_no_shutdown => 1, SSL_fast_shutdown => 1);
} else {
$socket->close;
}
};
return;
}
sub _slurpBinFile($self, $fname) {
# Read in file in binary mode, slurping it into a single scalar.
# We have to make sure we use binmode *and* turn on the line termination variable completly
# to work around the multiple idiosynchrasies of Perl on Windows
open(my $fh, "<", $fname) or croak($ERRNO);
local $INPUT_RECORD_SEPARATOR = undef;
binmode($fh);
my $data = <$fh>;
close($fh);
return $data;
}
sub _writeBinFile($self, $fname, $data) {
# Write file in binmode
# We have to make sure we use binmode *and* turn on the line termination variable completly
# to work around the multiple idiosynchrasies of Perl on Windows
open(my $fh, ">", $fname) or croak($ERRNO);
local $INPUT_RECORD_SEPARATOR = undef;
binmode($fh);
print $fh $data;
close($fh);
return true;
}
sub _restorePersistanceFile($self) {
my $previousfname = $self->{config}->{persistancefile} . '_bck';
my $tempfname = $self->{config}->{persistancefile} . '_';
my $loadok = 0;
if(-f $self->{config}->{persistancefile}) {
print "Trying to load persistance file ", $self->{config}->{persistancefile}, "\n";
$loadok = $self->_loadPersistanceFile($self->{config}->{persistancefile});
}
if(!$loadok && -f $previousfname) {
print "Trying to load backup (previous) persistance file ", $previousfname, "\n";
$loadok = $self->_loadPersistanceFile($previousfname);
if($loadok) {
$self->{savecache} = 2; # Force saving a new persistance file plus a new backup
}
}
if(!$loadok && -f $tempfname) {
print "Oh no. As a final, desperate solution, trying to load a 'temporary file while saving' persistance file ", $tempfname, "\n";
$loadok = $self->_loadPersistanceFile($tempfname);
if($loadok) {
$self->{savecache} = 2; # Force saving a new persistance file plus a new backup
}
}
if(!$loadok) {
print "Sorry, no valid persistance file found. Starting server 'blankety-blank'\n";
lib/Net/Clacks/Server.pm view on Meta::CPAN
next;
}
my $cachetime = $clackscachetime{$key};
my $accesstime = $now;
if(defined($clackscacheaccesstime{$key})) {
$accesstime = $clackscacheaccesstime{$key};
}
$cache{$key} = {
data => '',
cachetime => $cachetime,
accesstime => $accesstime,
deleted => 0,
};
}
my $converted = chr(0) . 'CLACKSV3' . Dump(\%cache) . chr(0) . 'CLACKSV3';
$self->_writeBinFile($fname, $converted);
print "...upgrade complete.\n";
return true;
}
sub _addInterclacksLink($self) {
my $now = $self->_getTime();
my $mcid;
if(defined($self->{config}->{master}->{socket})) {
$mcid = 'unixdomainsocket:interclacksmaster';
} else {
$mcid = $self->{config}->{master}->{ip}->[0] . ':' . $self->{config}->{master}->{port};
}
if(!defined($self->{clients}->{$mcid}) && $self->{nextinterclackscheck} < $now) {
$self->{nextinterclackscheck} = $now + $self->{config}->{interclacksreconnecttimeout} + int(rand(10));
print "Connect to master\n";
my $msocket;
if(defined($self->{config}->{master}->{socket})) {
$msocket = IO::Socket::UNIX->new(
Peer => $self->{config}->{master}->{socket}->[0],
Type => SOCK_STREAM,
);
} else {
$msocket = IO::Socket::IP->new(
PeerHost => $self->{config}->{master}->{ip}->[0],
PeerPort => $self->{config}->{master}->{port},
Type => SOCK_STREAM,
Timeout => 5,
);
}
if(!defined($msocket)) {
print STDERR "Can't connect to MASTER via interclacks!\n";
} else {
print "connected to master\n";
if(ref $msocket ne 'IO::Socket::UNIX') {
# ONLY USE SSL WHEN RUNNING OVER THE NETWORK
# There is simply no point in running it over a local socket.
my $encrypted = IO::Socket::SSL->start_SSL($msocket,
SSL_verify_mode => SSL_VERIFY_NONE,
);
if(!$encrypted) {
# Bare `next` was a bug here — there is no enclosing loop. Without an
# explicit close the failed-handshake $msocket leaks a file descriptor.
# Return cleanly; the next reconnect attempt will be scheduled by
# nextinterclackscheck below.
print STDERR "startSSL failed (interclacks master): ", $SSL_ERROR, "\n";
$self->_safeCloseSocket($msocket);
return;
}
}
$msocket->blocking(0);
#binmode($msocket, ':bytes');
my %tmp = (
buffer => '',
charbuffer => [],
listening => {},
socket => $msocket,
lastping => $now,
mirror => 0,
outbuffer => "CLACKS PageCamel $VERSION in interclacks client mode\r\n" . # Tell the server we are using PageCamel Interclacks...
"OVERHEAD A " . $self->{authtoken} . "\r\n" . # ...send Auth token
"OVERHEAD I 1\r\n", # ...and turn interclacks master mode ON on remote side
clientinfo => 'Interclacks link',
client_timeoffset => 0,
interclacks => 1,
interclacksclient => 1,
lastinterclacksping => $now,
lastmessage => $now,
authtimeout => $now + $self->{config}->{authtimeout},
authok => 0,
failtime => 0,
writefailtime => 0, # streak-start time for stalled writes; 0 = no streak in progress
outmessages => [],
inmessages => [],
messagedelay => 0,
inmessagedelay => 0,
outmessagedelay => 0,
permissions => {
read => 1,
write => 1,
manage => 1,
interclacks => 1,
},
);
if(defined($self->{config}->{master}->{ip})) {
$tmp{host} = $self->{config}->{master}->{ip}->[0];
$tmp{port} = $self->{config}->{master}->{port};
}
$self->{clients}->{$mcid} = \%tmp;
$msocket->_setClientID($mcid);
$self->{selector}->add($msocket);
$self->{workCount}++;
}
}
return;
}
sub _addNewClients($self) {
my $now = $self->_getTime();
foreach my $tcpsocket (@{$self->{tcpsockets}}) {
# Drain the listen queue rather than accepting one connection per iteration.
# The listener is non-blocking, so accept() returns undef once the queue is empty.
# This prevents a burst of connects from being rejected at the kernel level when
# the backlog fills up while the main loop is busy elsewhere.
my $acceptedThisCycle = 0;
while(1) {
# Hard cap to keep one rogue burst from monopolising the loop and starving
# existing clients. Any remaining pending connects will be picked up in the
# next runOnce() iteration.
last if($acceptedThisCycle >= 64);
my $clientsocket = $tcpsocket->accept;
if(!defined($clientsocket)) {
# EAGAIN/EWOULDBLOCK (no more pending) or a transient error like
# ECONNABORTED (peer RST between the kernel queueing the connection
# and our accept). Either way, stop draining this listener.
last;
}
$acceptedThisCycle++;
$clientsocket->blocking(0);
# Build a guaranteed-unique CID. The previous "$now:$rand(1_000_000)" scheme
# had a 50% birthday collision chance after ~1180 connects in the same
# second, and TCP CIDs based on $peerhost:$peerport collide whenever the
# kernel reuses an ephemeral port. A process-monotonic serial fixes both.
my $serial = $self->_nextClientSerial();
my ($cid, $chost, $cport);
if(ref $tcpsocket eq 'IO::Socket::UNIX') {
$chost = 'unixdomainsocket';
$cport = $now . ':' . $serial;
} else {
($chost, $cport) = ($clientsocket->peerhost, $clientsocket->peerport);
# Append the serial so a reused ephemeral port can never alias an
# existing or recently-disconnected client.
$cport .= ':' . $serial;
}
$cid = "$chost:$cport";
print "Got a new client $cid!\n";
foreach my $debugcid (keys %{$self->{clients}}) {
if($self->{clients}->{$debugcid}->{mirror}) {
$self->{clients}->{$debugcid}->{outbuffer} .= "DEBUG CONNECTED=" . $cid . "\r\n";
}
}
if(ref $clientsocket ne 'IO::Socket::UNIX') {
# ONLY USE SSL WHEN RUNNING OVER THE NETWORK
# There is simply no point in running it over a local socket.
my $encrypted = IO::Socket::SSL->start_SSL($clientsocket,
SSL_server => 1,
SSL_cert_file => $self->{config}->{ssl}->{cert},
SSL_key_file => $self->{config}->{ssl}->{key},
SSL_cipher_list => 'ALL:!ADH:!RC4:+HIGH:+MEDIUM:!LOW:!SSLv2:!SSLv3!EXPORT',
SSL_create_ctx_callback => sub {
my $ctx = shift;
# Enable workarounds for broken clients
Net::SSLeay::CTX_set_options($ctx, &Net::SSLeay::OP_ALL); ## no critic (Subroutines::ProhibitAmpersandSigils)
# Disable session resumption completely
Net::SSLeay::CTX_set_session_cache_mode($ctx, $SSL_SESS_CACHE_OFF);
# Disable session tickets
Net::SSLeay::CTX_set_options($ctx, &Net::SSLeay::OP_NO_TICKET); ## no critic (Subroutines::ProhibitAmpersandSigils)
},
);
if(!$encrypted) {
# Critical: must close the accepted FD here. Without this every
# failed handshake (port scan, RST during handshake, mismatched
# cert, slow client) leaks one OS file descriptor. After enough
# leaks the process hits RLIMIT_NOFILE and accept() begins
# returning undef forever, which looks like the server has hung.
print STDERR "startSSL failed for $cid: ", $SSL_ERROR, "\n";
$self->_safeCloseSocket($clientsocket);
next;
}
}
$clientsocket->blocking(0);
my %tmp = (
buffer => '',
charbuffer => [],
listening => {},
socket => $clientsocket,
lastping => $now,
mirror => 0,
outbuffer => "CLACKS PageCamel $VERSION\r\n" .
"OVERHEAD M Authentication required\r\n", # Informal message
clientinfo => 'UNKNOWN',
client_timeoffset => 0,
host => $chost,
port => $cport,
interclacks => 0,
interclacksclient => 0,
lastinterclacksping => 0,
lastmessage => $now,
authtimeout => $now + $self->{config}->{authtimeout},
authok => 0,
failtime => 0,
writefailtime => 0, # streak-start time for stalled writes; 0 = no streak in progress
outmessages => [],
inmessages => [],
inmessagedelay => 0,
outmessagedelay => 0,
permissions => {
read => 0,
write => 0,
manage => 0,
interclacks => 0,
( run in 0.958 second using v1.01-cache-2.11-cpan-0b5f733616e )