path: root/server/scireserver.pl

#!/usr/bin/perl

# $Id$

use strict;
use warnings;
use DBI;
use Data::Dumper;
use Digest::MD5 qw(md5 md5_hex );
use File::Path;
use Schedule::Cron::Events;


$| = 1;
$Data::Dumper::Purity = 1;

my $ETC_DIR = "/etc/scire";
my $SCIRE_CONFIG_FILE = "${ETC_DIR}/scireserver.conf";
my %conf;
my $LOGFILE;
	
my $conf_file = (defined($conf{config})) ? $conf{config} : $SCIRE_CONFIG_FILE;
read_config_file($conf_file);
Dumper(\%conf);

my $identified = 0;  #Global variable to determine if already identified or not.
my $client_id = 0;  #Clobal variable for the client id.
#  Somehow this feels insecure.

sub logger {
	my $line = shift;
	if(!defined $LOGFILE) {
		open(*LOGFILE, ">>$conf{logfile}") or die "Cannot open logfile $conf{logfile}";
	}
	print LOGFILE localtime() . " " . $line . "\n";
}

sub debug {
	my $line = shift;
	if ($conf{debug}) {
		if (defined($conf{logfile})) {
			logger("DEBUG: ${line}");
		} else {
			print STDERR "DEBUG: ${line}\n";
		}
	}
}

#Connect to the Database.
my $connect_string = "DBI:$conf{db_type}:$conf{db_name};host=$conf{db_host}";
debug("Connecting to $connect_string");
my $dbh = DBI->connect($connect_string, $conf{db_user}, $conf{db_passwd},  { RaiseError => 1 }  )
	or die "Could not connect to database: $DBI::errstr";

while(<>) {
	my ($command, @args) = parse_command($_);
#	chomp( my $line = $_);
	debug("DEBUG: line is: $_");
#	SEE http://agaffney.org/mediawiki/index.php/SSH-based_protocol for documentation on the protocol.

	if($command eq "QUIT") {
		print "OK\n";
		exit;
	}
		
	if($command eq "REGISTER") {
		my ($mac,$ip,$hostname) = @args;
		register_client($mac, $ip, $hostname);
		next; #End switch here.  You can go no further.
	}

	if($command eq "IDENTIFY") {
		my $fingerprint = $args[0];
		identify_client($fingerprint);
		next; #End switch here.  You can go no further.
	}
	unless($identified == 1) {
		print "ERROR This client has not yet been authorized.  Please identify!\n";
		next;
	}

	if ($command eq "GET_JOBS") {
		my @jobs = get_jobs();
		print "OK " . join(",", @jobs) . "\n";
	} elsif ($command eq "GET_JOB") {
		my $job = $args[0];
		my $jobfile = get_job($job);
		print "OK ${jobfile}\n";
	} elsif ($command eq "JOB_FETCHED") {
		my $job = $args[0];
		job_fetched($job) and print "OK\n";
	} elsif ($command eq "SET_JOB_STATUS") {
		my ($jobid,$status) = @args;
		set_job_status($jobid,$client_id,$status) and print "OK\n";
	} elsif ($command eq "RETURN_JOBFILE") {
		my $jobid = $args[0];
		my @filenames = ("$conf{job_dir}/$client_id/result/$jobid.stdout", "$conf{job_dir}/$client_id/result/$jobid.stderr");
		print "OK " . join(" ", @filenames) . "\n";
	} elsif ($command eq "JOBFILE_SENT") {
		my @jobfiles = @args;
		process_jobfile($_) foreach(@jobfiles);
		print "OK\n"
	} else {
		print "ERROR The command $command is unknown.  Please try again.\n";
	}
}



sub read_config_file {
	my $conf_file = shift;
	open(FH, "< ${conf_file}") or die("Couldn't open the config file ${conf_file}: $!");
	while (<FH>) {
		chomp;
		next if /^\s*(?:#|$)/;
		if(/^\s*(.+?)\s*=\s*(.+?)\s*(?:#.*)?$/) {
			unless(defined($conf{lc($1)})) {  #Don't overwrite anything specified in cmdline
				$conf{lc($1)} = $2;
			}
		}
	}
	close(FH) or die("Couldn't close the config file ${conf_file}: $!");
	debug("Conf file $conf_file read.");
}

#New clients must be registered so they can be given a key to use (perhaps for job file transfers?) for authentication.  This must be allowed before identifying.
sub register_client {
	my ($mac,$ip, $hostname) = @_;
	#Validate your inputs!
	$mac =~ /^[a-zA-Z0-9\:]+$/ or print "ERROR invalid mac $mac!\n";
	$ip =~ /^[a-zA-Z0-9\.\:]+$/ or print "ERROR invalid ip $ip!\n";

	my ($query, $status_id, $id, $sth);

	#Generate the digest
	my $digest = md5_hex(time()."${mac}${ip}${hostname}");
	
	eval {
		$query = 'SELECT statusid FROM client_status WHERE statusname = "Pending"';
		$sth = run_query($query);
		$status_id = $sth->fetchrow_hashref->{'statusid'};
    };
	($@) and print "ERROR Could not get status id: $DBI::errstr\n";

	eval {
		run_query('LOCK TABLES `gacl_axo_seq` WRITE');
		$query = 'SELECT id FROM `gacl_axo_seq`';
		$sth = run_query($query);
		$id = $sth->fetchrow_hashref->{'id'};
		$id += 1;
		$query = 'UPDATE `gacl_axo_seq` SET id=?';
		run_query($query,$id);
		run_query('UNLOCK TABLES');
	};
	($@) and print "ERROR during fetching of id sequence: $DBI::errstr\n";

	eval {
		$query = 'INSERT INTO `gacl_axo` (id,section_value,value,order_value,name,hidden) VALUES (?,"clients",?,"1",?,"0")'; 
		run_query($query,$id,$hostname,$hostname);
		#NOTE: not sure if this query is still valid.  may be using id instead of hostname for one of those two now.
		
		$query = 'INSERT INTO clients (clientid,digest,hostname,mac,ip,status) VALUES (?,?,?,?,?,?)';
		run_query($query,$id,$digest,$hostname,$mac,$ip,$status_id);
    };
	($@) and print "ERROR Could not insert client with $query: $DBI::errstr\n";
	#FIXME look for "duplicate key" and if found fail and notify admin.

	print "OK $digest\n";
}

#Identify the client by looking up the fingerprint in the database, and matching it up.
sub identify_client {
	my $digest = shift;
	#Validate your inputs!
	$digest =~ s/"//g;  #Clear the quotes.
	$digest =~ /^[A-Za-z0-9]+$/ or print "ERROR invalid digest!\n";

	my $query = 'SELECT client_status.statusname, clients.clientid FROM clients JOIN client_status on (clients.status = client_status.statusid) WHERE clients.digest=?';
	my $sth = run_query($query,$digest);
	my $hashref = $sth->fetchrow_hashref();
	debug(Dumper($hashref));
	my $status_name = $hashref->{'statusname'};
	$client_id = $hashref->{'clientid'};
	if (defined($client_id) and ($client_id > 0) and ($status_name eq 'Active')) {
		$identified = 1;
		print "OK\n";
	} else {
		print "ERROR Client could not be identified.  Status was $status_name\n";
	}

}

sub get_jobs {

	#FIXME expand jobs for $client_id
	expand_jobs();

	my $query = <<'EndOfQuery';
	SELECT jobs.jobid
	FROM jobs NATURAL JOIN jobs_clients NATURAL JOIN job_conditions
	WHERE jobs_clients.clientid = ? 
	AND jobs.jobid = jobs_clients.jobid 
	AND (job_conditions.deploy_time < now())
	AND ((job_conditions.expiration_time > now()) OR (job_conditions.expiration_time IS NULL))
	ORDER BY jobs.priority,jobs.created
EndOfQuery

	#FIXME ADD JOB DEPENDENCIES TO THIS QUERY.
	my $sth = run_query($query,$client_id);
	my $jobs_ref = $sth->fetchall_arrayref();
	# Don't ask me...ask the guys in #perl :P
	my @jobs = map { @$_ } @$jobs_ref;
	return @jobs;
}

sub get_job {
	my $jobid = shift;
	#Validate your inputs!
	my $query = 'SELECT * FROM jobs LEFT JOIN job_conditions on (jobs.jobid) WHERE jobs.jobid = ?';
	my $sth = run_query($query, $jobid);
	my $job = $sth->fetchrow_hashref();
	my $scriptid = $job->{'script'};

	$query = 'SELECT * FROM scripts WHERE scriptid=?';
	$sth = run_query($query,$scriptid);
	$job->{'script'} = $sth->fetchrow_hashref();
	
	debug(Dumper($job));
	#Write the job w/ all data to a jobfile with the following path /JOBDIR/CLIENT_ID/queue/JOBID.job
	my $path = "$conf{job_dir}/$client_id/queue";
	my $filename = "$path/$jobid.job";
	unless (-d $path) {
		print "WARNING! $path does not exist...creating\n";
		mkpath( $path, {verbose => 1, mode => 0660})
			or die("Couldn't make $path w/ perms 0660: $!");
	}
	open(FH, ">$filename") or die("Couldn't open $filename: $!");
	my $jobdata = Dumper($job);
	print FH $jobdata . "\n";
	close(FH) or die("Couldn't close $filename : $!");
	debug("OK $filename");
	return $filename;
}

sub job_fetched {
	my $jobid = shift;
	set_job_status($jobid,$client_id,'Downloaded', 'Job downloaded by client.') or print "ERROR could not set job status to downloaded.\n";

	eval {
		my $query = 'DELETE FROM jobs_clients WHERE jobid=? AND clientid=?';
		run_query($query,$jobid,$client_id);
	};
	($@) and print "ERROR Could not get status id: $DBI::errstr\n";

	my $filename = "$conf{job_dir}/$client_id/queue/$jobid.job";
	unlink ($filename) or die("ERROR Could not unlink the jobfile from the queue. filename: $filename : $!");
	return 1;
}

sub set_job_status {
	my ($jobid,$id_of_client,$status,$eventmsg) = @_;
	#Validate your inputs!
	$jobid =~ /^\d+$/ or die("Invalid jobid $jobid");
	$id_of_client ||= $client_id;
	$id_of_client =~ /^\d+$/ or die("Invalid id of client $id_of_client");
	$eventmsg ||= "Server status update.";
	#fixme validate status
	my $status_id;
	eval {
		my $query = 'SELECT statusid FROM jobs_status WHERE statusname = ?';
		my $sth = run_query($query,$status);
		$status_id = $sth->fetchrow_hashref->{'statusid'};
    };
	($@) and print "ERROR Could not get status id: $DBI::errstr\n";
	$status_id or print "ERROR Invalid status id $status_id\n";
	
	eval {
		my $query = 'INSERT INTO job_history (jobid,clientid,statusid,eventmsg) VALUES (?,?,?,?)';
		run_query($query,$jobid,$id_of_client,$status_id,$eventmsg);
	};
	($@) and print "ERROR Could not insert into job_history: $DBI::errstr\n";


	#If we're marking the completetion or failure of a job, we have more work to do here.
	if ($status eq 'Failed') {
		mark_job_as_failed($jobid,$id_of_client);
	} elsif ($status eq 'Completed') {
		mark_job_as_completed($jobid,$id_of_client);
	}

	return 1;
}

sub parse_command {
	my $line = shift;
	chomp $line;
	my @parts = split / (?!(?:[^" ]|[^"] [^"])+")/, $line;
	for(0..$#parts) {
		$parts[$_] =~ s/(^"|"$)//g;
		$parts[$_] =~ s/\\"/"/g;
	}
	return @parts;
}

sub run_query {
	my ($query, @params) = @_;
	debug("Query is $query");
	my $sth = $dbh->prepare($query);
	$sth->execute(@params);
	return $sth;
}


sub expand_jobs {
	#Searches for the group jobs that the client must be into and does the expansion.
	my @groups = get_client_groups();
	foreach my $groupid (@groups) {
		debug("Groupid is $groupid");
		my @members = get_group_clients($groupid);
		eval {
			my $query = <<'EndOfQuery2';
SELECT DISTINCT(jobs_clients.jobid)
FROM jobs_clients LEFT JOIN job_conditions on (jobs_clients.jobid=job_conditions.jobid)
WHERE jobs_clients.groupid = ? 
AND (job_conditions.deploy_time < now())
AND ((job_conditions.expiration_time > now()) OR (job_conditions.expiration_time IS NULL))
AND ((job_conditions.last_run_date < job_conditions.deploy_time) OR (job_conditions.last_run_date IS NULL))
EndOfQuery2
			my $sth = run_query($query,$groupid);
			run_query('LOCK TABLES `jobs_clients` WRITE, `job_conditions` WRITE, `job_history` WRITE, `jobs_status` WRITE, `jobs` WRITE');
			#FIXME need to lock jobs_clients for READ as well!!!
			while( my $jobref = $sth->fetchrow_hashref() ) {
				my $jobid = $jobref->{'jobid'};
				foreach my $member (@members) {
					$query = 'INSERT INTO jobs_clients (jobid, clientid) VALUES (?,?)';
					my $sth2 = run_query($query,$jobid,$member);
					
					set_job_status($jobid,$member,'Pending', 'Job expanded.') or print "ERROR could not add expanded jobs to job_history.\n";
				}
				$query = 'UPDATE `job_conditions` SET last_run_date = now() WHERE jobid = ?';
				run_query($query,$jobid);

				$query = 'UPDATE `jobs` SET pending=pending+? WHERE jobid = ?';
				run_query($query,$#members,$jobid);  #This works because you want one less b/c of removing the group.

				# One last query to remove the row from jobs_clients so someone else doesn't expand it.
				$query = 'DELETE FROM `jobs_clients` WHERE groupid=? AND jobid=?';
				run_query($query,$groupid,$jobid);

			}

			run_query('UNLOCK TABLES');
		};
		($@) and print "ERROR Could not expand jobs: $@  $DBI::errstr\n";
	return undef;
	}
}

sub mark_job_as_failed {
	my ($jobid,$id_of_client) = @_;
	
	#First off, update the pending count and failed count with the result.
	eval {
		my $query = 'UPDATE jobs SET pending=pending-1, failed=failed+1 WHERE jobid=?';
		run_query($query);
	};
	($@) and print "ERROR Could not update pending count: $@ $DBI::errstr\n";
}

sub mark_job_as_completed {
	my ($jobid,$id_of_client) = @_;
	my ($query,$sth);
	debug("Marking $jobid as completed for client $id_of_client");
	#If we succeeded, we need to check this jobid to see if it is a recurring job, and then set the next_run if necessary.
	#This requries looking at the pending count for the job as well as the run_schedule.
	
	#First off, update the pending count now that we've finished.
	eval {
		$query = 'UPDATE jobs SET pending=pending-1 WHERE jobid=?';
		run_query($query,$jobid);
	};
	($@) and print "ERROR Could not update pending count: $@ $DBI::errstr\n";
	
	#Now get the pending count and run_schedule.
	eval {
		$query = 'SELECT pending,run_schedule,expiration_time,deploy_time FROM jobs WHERE jobid=?';
		my $sth = run_query($query,$jobid);
		my $rowref = $sth->fetchrow_hashref();
		my ($sec,$min,$hour,$mday,$mon,$year,$wday,$yday,$isdst)=localtime(time);
		my $datetime = sprintf "%4d-%02d-%02d %02d:%02d:%02d\n",$year+1900,$mon+1,$mday,$hour,$min,$sec;
		if ($rowref->{'run_schedule'} and ($rowref->{'pending'} == 0) and ( $rowref->{'expiration_time'} > $datetime)) {
			#Determine the next run time.
			my $cron = new Schedule::Cron::Events( $rowref->{'run_schedule'},  Date => [ ( localtime(time()) )[0..5] ] );
			my ($sec, $min, $hour, $day, $month, $year) = $cron->nextEvent;
        	printf("Event will start next at %2d:%02d:%02d on %d %s, %d\n", $hour, $min, $sec, $day, $month, ($year+1900));
			
			#Get the groups and clients from the recurring_jobs_clients table.
			$query = 'SELECT clientid,groupid FROM recurring_jobs_clients WHERE jobid=?';
			my $sth2 = run_query($query,$jobid);
			while( my $recrow_ref = $sth2->fetchrow_hashref() ) {
				$query = 'INSERT INTO jobs_clients (jobid,clientid,groupid) VALUES (?,?,?)';
				run_query($query,$jobid,$recrow_ref->{'clientid'},$recrow_ref->{'groupid'});
			}
		}
		
	};
	($@) and print "ERROR Could not get run_schedule and pending count: $@ $DBI::errstr\n";
	
}

sub process_jobfile {
	my $filename = shift;

}

#########################################################
# PHPGACL FUNCTIONS
#########################################################

sub get_client_groups {
	my $query;
	my @groups;
	my $option = 'NO RECURSE';
	# If RECURSE it will get all ancestor groups.  defaults to only get direct parents.
	
	debug("get_object_groups(): Object ID: $client_id, option: $option");
	my $object_type = 'axo';
	my $group_table = 'gacl_axo_groups';
	my $map_table = 'gacl_groups_axo_map';

	if ($option eq 'RECURSE') {
		$query = "SELECT DISTINCT g.id as group_id FROM $map_table gm ";
		$query .= "LEFT JOIN $group_table g1 ON g1.id=gm.group_id ";
		$query .= "LEFT JOIN $group_table g ON g.lft<=g1.lft AND g.rgt>=g1.rgt";
	} else {
		$query = "SELECT gm.group_id FROM $map_table gm ";
	}
	$query .= " WHERE gm.axo_id=?";
	debug("Query is $query");
	eval {
		my $sth = $dbh->prepare($query);
		$sth->execute($client_id);
		my $groups_ref = $sth->fetchall_arrayref();
		# Don't ask me...ask the guys in #perl :P
		@groups = map { @$_ } @$groups_ref;
	};
	($@) and print "ERROR Could not get client groups: $DBI::errstr\n";
	return @groups;
}

sub get_group_clients {
	#This function gets the members of groups.  Returns an array containing those clients, empty otherwise.
	my $groupid = shift;
	my @members;
	my $query = 'SELECT axo_id FROM gacl_groups_axo_map WHERE group_id = ?';
	debug("Query is $query");
	eval {
		my $sth = $dbh->prepare($query);
		$sth->execute($groupid);
		my $members_ref = $sth->fetchall_arrayref();
		# Don't ask me...ask the guys in #perl :P
		@members = map { @$_ } @$members_ref;
	};
	($@) and print "ERROR Could not get group members: $DBI::errstr\n";
	return @members;
}