James Labocki’s Blog

Here are the steps I took to audit database activity and get the data into Splunk. You could technically send this to any log repository capable of listening on a TCP socket. The challenge is that many RDBMS don’t allow you to get database information. Oracle is notorious for this (because they want you to purchase Database Vault). Of course if you don’t have money (like me) for that you can use this solution. You really need three things:

1. Install DBI and Oracle:DBD on the host that will be polling the database.
2. Grant a user who will run the script SELECT on dba_audit_trail table (or whatever table you want to poll and pull into splunk).
3. Setup the perl script with the proper inputs and encrypt the password file for no plain text on the file system.

1. Install DBD::Oracle on 64-bit Linux and Oracle 11g

DBD::Oracle 1.21 to install on a 64-bit Linux OS against the Oracle 11 full client. As root, we downloaded DBD::Oracle from CPAN.


# perl -MCPAN -eshell
cpan> install DBI
cpan> get DBD::Oracle
...
# cd /root/.cpan/build/DBD-Oracle-1.21
# export ORACLE_HOME=actual value of Oracle Home
# export ORACLE_SID=actual value of ORACLE_SID
# export ORACLE_USERID=working ORACLE_USERID
# export LD_LIBRARY_PATH=$ORACLE_HOME/lib:$ORACLE_HOME
# perl Makefile.PL
...
# make
...
# make test
...
# make install


Now you can install the dbipoll perl script.

Generate a random key, encode a file with it, and then decode a file with it. Useful to generate a password, encrypt it and then get rid of the plain text password inside scripts.

openssl rand 32 -out key.temp
openssl des3 -e -pass file:key.temp -in passwd -out passwd.enc
openssl des3 -d pass file:key.dec -in passwd.enc -out passwd.dec
openssl des3 -d -pass file:key.dec -in passwd.enc -out passwd.dec
openssl des3 -d -pass file:key.temp -in passwd.enc -out passwd.dec

Verify you are running encrypted. Add the following temporarily to your sqlnet.ora

TRACE_DIRECTORY_CLIENT=/ora-main/app/oracle/product/11.1.0.6/db/network/trace
#TRACE_FILE_CLIENT=client
TRACE_LEVEL_CLIENT=16

Inspect the trace for encryption

[root@mymachine trace]# cat cli_9668.trc |grep encry
(2863411744) [13-FEB-2009 15:37:50:911] naeshow: These are the encryption algorithms that the client will accept:
(2863411744) [13-FEB-2009 15:37:50:911] naeshow: Choice 0: no algorithm; encryption inactive
service: encryption
(2863411744) [13-FEB-2009 15:37:50:934] na_info: Oracle Advanced Security: encryption service for Linux: Version 11.1.0.6.0 - Production
service: encryption
(2863411744) [13-FEB-2009 15:37:50:935] nacomps: service encryption
(2863411744) [13-FEB-2009 15:37:50:940] nacomus: service encryption
service: encryption
service: encryption
(2863411744) [13-FEB-2009 15:37:50:952] naeecom: The server chose the 'RC4_256' encryption algorithm
(2863411744) [13-FEB-2009 15:37:50:952] na_tns: encryption is active, using RC4_256

Note the last two lines indicating we are encrypted.

Here is the dbipoll.pl script.


#!/usr/bin/perl
#
#
# Now accepts arguments for countkey, table, and columns.
# countkey will ideally be a timestamp field, such as EXTENDED_TIMESTAMP in AUD$.
#
# Need to deal with the following variations...
# 1 - Oracle local
# 2 - Oracle remote
# 3 - mysql local
# 4 - mysql remote
# 5 - Sybase local
# 6 - Sybase remote
#
# Oracle connect string: $dbh = DBI->connect("dbi:Oracle:host=$server;sid=$db;port=$dbport", $user, $passwd);
# Oracle connect string without env vars: $dbh = DBI->connect( "dbi:Oracle:$db", $username, $passwd )
# mysql connect string: $dbh = DBI->connect("dbi:mysql:database=$db\;host=$server;port=$dbport","$user","$passwd");
# Sybase connect string: $dbh = DBI->connect("dbi:sybase", $username, $password, $db);
use IO::Socket;
use strict;
use DBI;
use Getopt::Long;
use Sys::Syslog;
#
our ($server,$db,$table,$columns,$user,$enpasswd,$passwd,$dbtype,$tcphost,$tcpport,
$countfile,$countkey,$ofile,$dbport,$baseconnect,$dbh);
#
GetOptions(
"dbtype=s" =>\$dbtype, # Type of DB (Oracle, MySQL, etc.)
"server=s" =>\$server, # IP address of database server.
"dbport=s" =>\$dbport, # Port for connecting to remote DB.
"db=s" =>\$db, # Database to retriever records from.
"user=s" =>\$user, # DB username.
"password=s" =>\$passwd, # Unencrypted password.
"xpassword=s" =>\$enpasswd, # Encrypted password.
"table=s" =>\$table, # Table to retrieve from.
"columns=s" =>\$columns, # Columns to fetch.
"tcphost=s" =>\$tcphost, # Splunk server IP.
"tcpport=s" =>\$tcpport, # Splunk tcp listener port.
"countfile=s" =>\$countfile, # File to store row count status.
"countkey=s" =>\$countkey, # Field to use as count value
"o=s" =>\$ofile, # File for output (default is stdout).
);
#
# Deal with encrypted passwords first...
if (!$passwd) {
if ($enpasswd) {
$passwd = `openssl des3 -d -pass file:/etc/pki/tls/private/dbi.key -in $enpasswd |awk NF`;
chomp($passwd);
}
}
#
if ($dbtype =~ /oracle/i) {
$ENV{'DBI_DRIVER'} = "Oracle";
$dbtype = "Oracle";
$dbh->{InactiveDestroy} = 1;
#$baseconnect = DBI->connect("dbi:$dbtype:$db");
if (!$dbport) {
# Assume a local DB.
$dbh = DBI->connect("dbi:$dbtype:$db", "$user", "$passwd") ||
die( $DBI::errstr . "\n" );
#
} else {
if (!$passwd) {
$dbh = DBI->connect("dbi:Oracle:host=$server;sid=$db;port=$dbport", $user) ||
die( $DBI::errstr . "\n" );
} else {
$dbh = DBI->connect("dbi:Oracle:host=$server;sid=$db;port=$dbport", $user, $passwd) ||
die( $DBI::errstr . "\n" );
}
}
#
}
if ($dbtype =~ /mysql/i) {
$ENV{'DBI_DRIVER'} = "mysql";
$dbtype = "mysql";
#$baseconnect = DBI->connect("dbi:$dbtype:database=$db;host=$server");
if (!$dbport) {
# Assume a local DB.
$dbh = DBI->connect("dbi:$dbtype:host=$server;database=$db","$user","$passwd") ||
die( $DBI::errstr . "\n" );
#$dbh = $baseconnect . "$user" . "$passwd";
} else {
if (!$passwd) {
$dbh = DBI->connect("dbi:$dbtype:database=$db;host=$server;port=$dbport","$user") ||
die( $DBI::errstr . "\n" );
} else {
$dbh = DBI->connect("dbi:$dbtype:database=$db;host=$server;port=$dbport","$user","$passwd") ||
die( $DBI::errstr . "\n" );
}
}
#
}
if (($dbtype =~ /sybase/i) && (!$ENV{'SYBASE'})) {
$ENV{'SYBASE'} = '/usr/local/freetds';
$ENV{'DSQUERY'} = "$server";
$dbtype = "sybase";
#$baseconnect = DBI->connect("dbi:$dbtype", "$db");
if (!$dbport) {
# Assume a local DB.
$dbh = DBI->connect("dbi:$dbtype:database=$db","$user","$passwd") ||
die( $DBI::errstr . "\n" );
} else {
if (!$passwd) {
$dbh = DBI->connect("dbi:$dbtype:database=$db;port=$dbport","$user") ||
die( $DBI::errstr . "\n" );
} else {
$dbh = DBI->connect("dbi:$dbtype:database=$db;port=$dbport","$user","$passwd") ||
die( $DBI::errstr . "\n" );
}
}
}
my $sth;
if ( -f $countfile) {
# Compare vs. file with saved count
open (CF, "<$countfile");
my $filecount = readline CF;
chomp $filecount;
close (CF);
# Adding an ORDER BY clause here may be a requirement
$sth = $dbh->prepare("SELECT $columns FROM $table where $countkey > \'$filecount\' order by $countkey");
} else {
$sth = $dbh->prepare("SELECT $columns FROM $table order by $countkey");
}
#
our $handle;
if (($tcphost) && ($tcpport)) {
# Send directly to splunk server...
# Create tcp socket to send the data to the splunk server
my $remote = IO::Socket::INET->new( Proto => "tcp",
PeerAddr => $tcphost,
PeerPort => $tcpport,
Type => SOCK_STREAM)
or die "cannot connect to tcp daemon on $tcphost";
$remote->autoflush(1);
$handle = $remote;
#$target = "\$remote";
} elsif ($ofile) {
# Write to local outfile...
open (OFILE, ">$ofile");
$handle = *OFILE;
#$target = *OFILE;
} else {
# send to stdout
open (STDOUT, ">&1");
$handle = *STDOUT;
#$target = *STDOUT;
}
#
# Get some Rows...
$sth->execute or die $sth->errstr;
open (CF, ">$countfile");
while(my $hash_ref = $sth->fetchrow_hashref) {
my $output = "";
my $lastrow;
foreach my $key (keys(%$hash_ref)) {
my $str = $hash_ref->{$key};
if ($key ne "EXTENDED_TIMESTAMP") {
if ($str =~ /"/) {
(my $xstr = $str) =~ s/"/\\"/g;
$str = "\"$xstr\"";
} else {
$str = "\"$str\"" if ($str =~ / /);
$str = "\"$str\"" if ($str =~ /,/);
}
$output = $output . $key . "=" . $str . ",";
}
if ($key eq "EXTENDED_TIMESTAMP") {
# if ($key eq "TO_CHAR(EXTENDED_TIMESTAMP,'MM/DD/YYYYHH24:MI:SS')") {
$str =~ s/\"//g;
$lastrow = $str; #if (lc($str) eq lc($countkey));
}
#
}
chop($output);
# Update countfile
# Where did we leave off?...
*CF->autoflush(1);
seek(CF,0,0);
print CF "$lastrow\n";
#
# And send them somewhere.
print $handle $output . "\n";
}
#
$sth->finish();
$dbh->disconnect;
#
close (CF);
close $handle;


Now schedule this to run (I just wrote a wrapper script in ksh exporting the variables to be safe).


#!/usr/bin/ksh
#
export ORACLE_HOME=/ora-main/app/oracle/product/11.1.0.6/db
export ORACLE_SID=rmdev1
export LD_LIBRARY_PATH=$ORACLE_HOME/lib:$ORACLE_HOME
#
/usr/bin/dbipoll.pl -dbtype=oracle -server=servername -dbport=1521 -db=development -user=splunk -xpassword=/etc/dbipasswd -table=dba_audit_trail -columns="OS_PROCESS,OS_USERNAME,USERNAME,USERHOST,ACTION,ACTION_NAME,TRANSACTIONID,SCN,SQL_BIND,SQL_TEXT,TO_CHAR(EXTENDED_TIMESTAMP,'MM/DD/YYYY HH24:MI:SS'),EXTENDED_TIMESTAMP" -countfile="/app/oracle/product/monitor/countfile" -countkey="EXTENDED_TIMESTAMP" -tcphost=splunkserver -tcpport=9094


Of course you will also need to setup a tcp input port on your splunk server. Once you have all that in place you can run the following query to see what hosts are performing the largest amount of actions on the database and what users are running the majority of actions on the database. I have removed hostnames and usernames legends from the following for security.

index=indexname sourcetype=databasesource | timechart count(USERNAME) by USERNAME

index=indexname sourcetype=databasesource |timechart count(USERHOST) by USERHOST

You can also use this to setup alerts if any users runs an update on a database (useful to catch those sneaky developers).