summaryrefslogtreecommitdiff
blob: 1af39ee0885c2dc9f3665ef934b830f4f5e615bc (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
#!/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;
}