File Coverage

blib/lib/Net/MySQL.pm

Criterion	Covered	Total	%
statement	226	526	42.9
branch	11	108	10.1
condition	3	15	20.0
subroutine	60	108	55.5
pod	11	13	84.6
total	311	770	40.3

package Net::MySQL;

use 5.004;
use IO::Socket;
use Carp;
use vars qw($VERSION $DEBUG);
use strict;
$VERSION = '0.11';

use constant COMMAND_SLEEP          => "\x00";
use constant COMMAND_QUIT           => "\x01";
use constant COMMAND_INIT_DB        => "\x02";
use constant COMMAND_QUERY          => "\x03";
use constant COMMAND_FIELD_LIST     => "\x04";
use constant COMMAND_CREATE_DB      => "\x05";
use constant COMMAND_DROP_DB        => "\x06";
use constant COMMAND_REFRESH        => "\x07";
use constant COMMAND_SHUTDOWN       => "\x08";
use constant COMMAND_STATISTICS     => "\x09";
use constant COMMAND_PROCESS_INFO   => "\x0A";
use constant COMMAND_CONNECT        => "\x0B";
use constant COMMAND_PROCESS_KILL   => "\x0C";
use constant COMMAND_DEBUG          => "\x0D";
use constant COMMAND_PING           => "\x0E";
use constant COMMAND_TIME           => "\x0F";
use constant COMMAND_DELAYED_INSERT => "\x10";
use constant COMMAND_CHANGE_USER    => "\x11";
use constant COMMAND_BINLOG_DUMP    => "\x12";
use constant COMMAND_TABLE_DUMP     => "\x13";
use constant COMMAND_CONNECT_OUT    => "\x14";

use constant DEFAULT_PORT_NUMBER => 3306;
use constant BUFFER_LENGTH       => 1460;
use constant DEFAULT_UNIX_SOCKET => '/tmp/mysql.sock';


sub new
{
	my $class = shift;
	my %args = @_;

	my $self = bless {
		hostname   => $args{hostname},
		unixsocket => $args{unixsocket} || DEFAULT_UNIX_SOCKET,
		port       => $args{port}       || DEFAULT_PORT_NUMBER,
		database   => $args{database},
		user       => $args{user},
		password   => $args{password},
		timeout    => $args{timeout}  || 60,
		'socket'   => undef,
		salt                 => '',
		protocol_version     => undef,
		client_capabilities  => 0,
		affected_rows_length => 0,
	}, $class;
	$self->debug($args{debug});
	$self->_initialize;
	return $self;
}


sub query
{
	my $self = shift;
	my $sql = join '', @_;
	my $mysql = $self->{socket};

	return $self->_execute_command(COMMAND_QUERY, $sql);
}


sub create_database
{
	my $self = shift;
	my $db_name = shift;
	my $mysql = $self->{socket};

	return $self->_execute_command(COMMAND_CREATE_DB, $db_name);
}


sub drop_database
{
	my $self = shift;
	my $db_name = shift;
	my $mysql = $self->{socket};

	return $self->_execute_command(COMMAND_DROP_DB, $db_name);
}


sub close
{
	my $self = shift;
	my $mysql = $self->{socket};
	return unless $mysql->can('send');

	my $quit_message =
		chr(length(COMMAND_QUIT)). "\x00\x00\x00". COMMAND_QUIT;
	$mysql->send($quit_message, 0);
	$self->_dump_packet($quit_message) if Net::MySQL->debug;
	$mysql->close;
}


sub get_affected_rows_length
{
	my $self = shift;
	$self->{affected_rows_length};
}


sub get_insert_id
{
	my $self = shift;
	$self->{insert_id};
}


sub create_record_iterator
{
	my $self = shift;
	return undef unless $self->has_selected_record;

	my $record = Net::MySQL::RecordIterator->new(
		$self->{selected_record}
	);
	$self->{selected_record} = undef;
	$record->parse;
	return $record;
}


sub has_selected_record
{
	my $self = shift;
	$self->{selected_record} ? 1 : undef;
}


sub is_error
{
	my $self = shift;
	$self->{error_code} ? 1 : undef;
}


sub get_error_code
{
	my $self = shift;
	$self->{error_code};
}


sub get_error_message
{
	my $self = shift;
	$self->{server_message};
}


sub debug
{
	my $class = shift;
	$DEBUG = shift if @_;
	$DEBUG;
}


sub _connect
{
	my $self = shift;

	my $mysql;
	if ($self->{hostname}) {
		printf "Use INET Socket: %s %d/tcp\n", $self->{hostname}, $self->{port}
			if $self->debug;
		$mysql = IO::Socket::INET->new(
			PeerAddr => $self->{hostname},
			PeerPort => $self->{port},
			Proto    => 'tcp',
			Timeout  => $self->{timeout} || 60,
		) or croak "Couldn't connect to $self->{hostname}:$self->{port}/tcp: $@";
	}
	else {
		printf "Use UNIX Socket: %s\n", $self->{unixsocket} if $self->debug;
		$mysql = IO::Socket::UNIX->new(
			Type => SOCK_STREAM,
			Peer => $self->{unixsocket},
		) or croak "Couldn't connect to $self->{unixsocket}: $@";
	}
	$mysql->autoflush(1);
	$self->{socket} = $mysql;
}


sub _get_server_information
{
	my $self = shift;
	my $mysql = $self->{socket};

	my $message;
	$mysql->recv($message, BUFFER_LENGTH, 0);
	$self->_dump_packet($message)
		if Net::MySQL->debug;
	my $i = 0;
	my $packet_length = ord substr $message, $i, 1;
	$i += 4;
	$self->{protocol_version} = ord substr $message, $i, 1;
	printf "Protocol Version: %d\n", $self->{protocol_version}
		if Net::MySQL->debug;
	if ($self->{protocol_version} == 10) {
		$self->{client_capabilities} = 1;
	}

	++$i;
	my $string_end = index($message, "\0", $i) - $i;
	$self->{server_version} = substr $message, $i, $string_end;
	printf "Server Version: %s\n", $self->{server_version}
		if Net::MySQL->debug;

	$i += $string_end + 1;
	$self->{server_thread_id} = unpack 'v', substr $message, $i, 2;
	$i += 4;
	$self->{salt} = substr $message, $i, 8;
	#
	$i += 8+1;
	if (length $message >= $i + 1) {
		$i += 1;
	}
	if (length $message >= $i + 18) {
		# get server_language
		# get server_status
	}
	$i += 18 - 1;
	if (length $message >= $i + 12 - 1) {
		$self->{salt} .= substr $message, $i, 12;
	}
	printf "Salt: %s\n", $self->{salt} if Net::MySQL->debug;

}


sub _request_authentication
{
	my $self = shift;
	my $mysql = $self->{socket};
	$self->_send_login_message();

	my $auth_result;
	$mysql->recv($auth_result, BUFFER_LENGTH, 0);
	$self->_dump_packet($auth_result) if Net::MySQL->debug;
	if ($self->_is_error($auth_result)) {
		$mysql->close;
		if (length $auth_result < 7) {
			croak "Timeout of authentication";
		}
		croak substr $auth_result, 7;
	}
	print "connect database\n" if Net::MySQL->debug;
}


sub _send_login_message
{
	my $self = shift;
	my $mysql = $self->{socket};
	my $body = "\0\0\x01\x0d\xa6\03\0\0\0\0\x01".
		"\x21\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0".
		 join "\0",
		$self->{user},
		"\x14".
		Net::MySQL::Password->scramble(
			$self->{password}, $self->{salt}, $self->{client_capabilities}
		);
	$body .= $self->{database};
	$body .= "\0";
	my $login_message = chr(length($body)-3). $body;
	$mysql->send($login_message, 0);
	$self->_dump_packet($login_message) if Net::MySQL->debug;
}



sub _execute_command
{
	my $self = shift;
	my $command = shift;
	my $sql = shift;
	my $mysql = $self->{socket};

	my $message = pack('V', length($sql) + 1). $command. $sql;
	$mysql->send($message, 0);
	$self->_dump_packet($message) if Net::MySQL->debug;

	my $result;
	$mysql->recv($result, BUFFER_LENGTH, 0);
	$self->_dump_packet($result) if Net::MySQL->debug;
	$self->_reset_status;

	if ($self->_is_error($result)) {
		return $self->_set_error_by_packet($result);
	}
	elsif ($self->_is_select_query_result($result)) {
		return $self->_get_record_by_server($result);
	}
	elsif ($self->_is_update_query_result($result)){
		return $self->_get_affected_rows_information_by_packet($result);
	}
	else {
		croak 'Unknown Result: '. $self->_get_result_length($result). 'byte';
	}
}


sub _initialize
{
	my $self = shift;
	$self->_connect;
	$self->_get_server_information;
	$self->_request_authentication;
}


sub _set_error_by_packet
{
	my $self = shift;
	my $packet = shift;

	my $error_message = $self->_get_server_message($packet);
	$self->{server_message} = $error_message;
	$self->{error_code}     = $self->_get_error_code($packet);
	return undef;
}


sub _get_record_by_server
{
	my $self = shift;
	my $packet = shift;
	my $mysql = $self->{socket};
	$self->_get_column_length($packet);
	while ($self->_has_next_packet($packet)) {
		my $next_result;
		$mysql->recv($next_result, BUFFER_LENGTH, 0);
		$packet .= $next_result;
		$self->_dump_packet($next_result) if Net::MySQL->debug;
	}
	$self->{selected_record} = $packet;
}


sub _get_affected_rows_information_by_packet
{
	my $self = shift;
	my $packet = shift;

	$self->{affected_rows_length} = $self->_get_affected_rows_length($packet);
	$self->{insert_id} = $self->_get_insert_id($packet);
	$self->{server_message} = $self->_get_server_message($packet);
	return $self->{affected_rows_length};
}


sub _is_error
{
	my $self = shift;
	my $packet = shift;
	return 1 if length $packet < 4;
	ord(substr $packet, 4) == 255;
}


sub _is_select_query_result
{
	my $self = shift;
	my $packet = shift;
	return undef if $self->_is_error($packet);
	ord(substr $packet, 4) >= 1;
}


sub _is_update_query_result
{
	my $self = shift;
	my $packet = shift;
	return undef if $self->_is_error($packet);
	ord(substr $packet, 4) == 0;
}


sub _get_result_length
{
	my $self = shift;
	my $packet = shift;
	ord(substr $packet, 0, 1)
}


sub _get_column_length
{
	my $self = shift;
	my $packet = shift;
	ord(substr $packet, 4);
}


sub _get_affected_rows_length
{
	my $self = shift;
	my $packet = shift;
	my $pos = 5;
	return Net::MySQL::Util::get_field_length($packet, \$pos);
}


sub _get_insert_id
{
	my $self = shift;
	my $packet = shift;
	return ord(substr $packet, 6, 1) if ord(substr $packet, 6, 1) != 0xfc;
	unpack 'v', substr $packet, 7, 2;
}


sub _get_server_message
{
	my $self = shift;
	my $packet = shift;
	return '' if length $packet < 7;
	substr $packet, 7;
}


sub _get_error_code
{
	my $self = shift;
	my $packet = shift;
	$self->_is_error($packet)
		or croak "_get_error_code(): Is not error packet";
	unpack 'v', substr $packet, 5, 2;
}


sub _reset_status
{
	my $self = shift;
	$self->{insert_id}       = 0;
	$self->{server_message}  = '';
	$self->{error_code}      = undef;
	$self->{selected_record} = undef;
}


sub _has_next_packet
{
	my $self = shift;
	#substr($_[0], -1) ne "\xfe";
	#$self->_dump_packet(substr($_[0], -5));
	return substr($_[0], -5, 1) ne "\xfe";
}


sub _dump_packet {
    my $self = shift;
    my $packet = shift;
    my ($method_name) = (caller(1))[3];
    my $str = sprintf "%s():\n", $method_name;
    while ($packet =~ /(.{1,16})/sg) {
        my $line = $1;
        $str .= join ' ', map {sprintf '%02X', ord $_} split //, $line;
        $str .= '   ' x (16 - length $line);
        $str .= '  ';
        $str .= join '', map {
            sprintf '%s', (/[\w\d\*\,\?\%\=\'\;\(\)\.-]/) ? $_ : '.'
        } split //, $line;
        $str .= "\n";
    }
    print $str;
    #warn $str;
}



package Net::MySQL::RecordIterator;
use strict;

use constant NULL_COLUMN           => 251;
use constant UNSIGNED_CHAR_COLUMN  => 251;
use constant UNSIGNED_SHORT_COLUMN => 252;
use constant UNSIGNED_INT24_COLUMN => 253;
use constant UNSIGNED_INT32_COLUMN => 254;
use constant UNSIGNED_CHAR_LENGTH  => 1;
use constant UNSIGNED_SHORT_LENGTH => 2;
use constant UNSIGNED_INT24_LENGTH => 3;
use constant UNSIGNED_INT32_LENGTH => 4;
use constant UNSIGNED_INT32_PAD_LENGTH => 4;


sub new
{
	my $class = shift;
	my $packet = shift;
	bless {
		packet   => $packet,
		position => 0,
		column   => [],
	}, $class;
}


sub parse
{
	my $self = shift;
	$self->_get_column_length;
	$self->_get_column_name;
}


sub each
{
	my $self = shift;
	my @result;
	return undef if $self->is_end_of_packet;

	for (1..$self->{column_length}) {
		push @result, $self->_get_string_and_seek_position;
	}
	$self->{position} += 4;

	return \@result;
}


sub is_end_of_packet
{
	my $self = shift;
	return substr($self->{packet}, $self->{position}, 1) eq "\xFE";
}


sub get_field_length
{
	my $self = shift;
	$self->{column_length};
}


sub get_field_names
{
	my $self = shift;
	map { $_->{column} } @{$self->{column}};
}


sub _get_column_length
{
	my $self = shift;
	$self->{position} += 4;
	$self->{column_length} = ord substr $self->{packet}, $self->{position}, 1;
	$self->{position} += 5;
	printf "Column Length: %d\n", $self->{column_length}
		if Net::MySQL->debug;
}


sub _get_column_name
{
	my $self = shift;

	for my $i (1.. $self->{column_length}) {
		$self->_get_string_and_seek_position;
		$self->_get_string_and_seek_position;
		my $table = $self->_get_string_and_seek_position;
		$self->_get_string_and_seek_position;
		my $column = $self->_get_string_and_seek_position;
		$self->_get_string_and_seek_position;
		push @{$self->{column}}, {
			table  => $table,
			column => $column,
		};
		$self->_get_string_and_seek_position;
		$self->{position} += 4;
	}
	$self->{position} += 9;
	printf "Column name: '%s'\n",
		join ", ", map { $_->{column} } @{$self->{column}}
			if Net::MySQL->debug;
}


sub _get_string_and_seek_position
{
	my $self = shift;

	my $length = $self->_get_field_length();

	return undef unless defined $length;

	my $string = substr $self->{packet}, $self->{position}, $length;
	$self->{position} += $length;
	return $string;
}


sub _get_field_length
{
	my $self = shift;
	return Net::MySQL::Util::get_field_length($self->{packet}, \$self->{position});
}


package Net::MySQL::Util;
use strict;

use constant NULL_COLUMN           => 251;
use constant UNSIGNED_CHAR_COLUMN  => 251;
use constant UNSIGNED_SHORT_COLUMN => 252;
use constant UNSIGNED_INT24_COLUMN => 253;
use constant UNSIGNED_INT32_COLUMN => 254;
use constant UNSIGNED_CHAR_LENGTH  => 1;
use constant UNSIGNED_SHORT_LENGTH => 2;
use constant UNSIGNED_INT24_LENGTH => 3;
use constant UNSIGNED_INT32_LENGTH => 4;
use constant UNSIGNED_INT32_PAD_LENGTH => 4;


sub get_field_length
{
	my $packet = shift;
	my $pos = shift;

	my $head = ord substr(
		$packet,
		$$pos,
		UNSIGNED_CHAR_LENGTH
	);
	$$pos += UNSIGNED_CHAR_LENGTH;

	return undef if $head == NULL_COLUMN;
	if ($head < UNSIGNED_CHAR_COLUMN) {
		return $head;
	}
	elsif ($head == UNSIGNED_SHORT_COLUMN) {
		my $length = unpack 'v', substr(
			$packet,
			$$pos,
			UNSIGNED_SHORT_LENGTH
		);
		$$pos += UNSIGNED_SHORT_LENGTH;
		return $length;
	}
	elsif ($head == UNSIGNED_INT24_COLUMN) {
		my $int24 = substr(
			$packet, $$pos,
			UNSIGNED_INT24_LENGTH
		);
		my $length = unpack('C', substr($int24, 0, 1))
		          + (unpack('C', substr($int24, 1, 1)) << 8)
			  + (unpack('C', substr($int24, 2, 1)) << 16);
		$$pos += UNSIGNED_INT24_LENGTH;
		return $length;
	}
	else {
		my $int32 = substr(
			$packet, $$pos,
			UNSIGNED_INT32_LENGTH
		);
		my $length = unpack('C', substr($int32, 0, 1))
		          + (unpack('C', substr($int32, 1, 1)) << 8)
			  + (unpack('C', substr($int32, 2, 1)) << 16)
			  + (unpack('C', substr($int32, 3, 1)) << 24);
		$$pos += UNSIGNED_INT32_LENGTH;
		$$pos += UNSIGNED_INT32_PAD_LENGTH;
		return $length;
	}
}



package Net::MySQL::Password;
use strict;
use Digest::SHA1;

sub scramble {
	my $class = shift;
	my $password = shift;
	my $hash_seed = shift;
	return '' unless $password;
	return '' if length $password == 0;
	return _make_scrambled_password($hash_seed, $password);
}


sub _make_scrambled_password {
	my $message = shift;
	my $password = shift;

	my $ctx = Digest::SHA1->new;
	$ctx->reset;
	$ctx->add($password);
	my $stage1 = $ctx->digest;

	$ctx->reset;
	$ctx->add($stage1);
	my $stage2 = $ctx->digest;

	$ctx->reset;
	$ctx->add($message);
	$ctx->add($stage2);
	my $result = $ctx->digest;
	return _my_crypt($result, $stage1);
}

sub _my_crypt {
	my $s1 = shift;
	my $s2 = shift;
	my $l = length($s1) - 1;
	my $result = '';
	for my $i (0..$l) {
		$result .= pack 'C', (unpack('C', substr($s1, $i, 1)) ^ unpack('C', substr($s2, $i, 1)));
	}
	return $result;
}

package Net::MySQL::Password32;
use strict;

sub scramble
{
	my $class = shift;
	my $password = shift;
	my $hash_seed = shift;
	my $client_capabilities = shift;

	return '' unless $password;
	return '' if length $password == 0;

	my $hsl = length $hash_seed;
	my @out;
	my @hash_pass = _get_hash($password);
	my @hash_mess = _get_hash($hash_seed);

	my ($max_value, $seed, $seed2);
	my ($dRes, $dSeed, $dMax);
	if ($client_capabilities < 1) {
		$max_value = 0x01FFFFFF;
		$seed = _xor_by_long($hash_pass[0], $hash_mess[0]) % $max_value;
		$seed2 = int($seed / 2);
	} else {
		$max_value= 0x3FFFFFFF;
		$seed  = _xor_by_long($hash_pass[0], $hash_mess[0]) % $max_value;
		$seed2 = _xor_by_long($hash_pass[1], $hash_mess[1]) % $max_value;
	}
	$dMax = $max_value;

	for (my $i=0; $i < $hsl; $i++) {
		$seed  = int(($seed * 3 + $seed2) % $max_value);
		$seed2 = int(($seed + $seed2 + 33) % $max_value);
		$dSeed = $seed;
		$dRes = $dSeed / $dMax;
		push @out, int($dRes * 31) + 64;
	}

	if ($client_capabilities == 1) {
		# Make it harder to break
		$seed  = ($seed * 3 + $seed2  ) % $max_value;
		$seed2 = ($seed + $seed2 + 33 ) % $max_value;
		$dSeed = $seed;

		$dRes = $dSeed / $dMax;
		my $e = int($dRes * 31);
		for (my $i=0; $i < $hsl ; $i++) {
			$out[$i] ^= $e;
		}
	}
	return join '', map { chr $_ } @out;
}


sub _get_hash
{
	my $password = shift;

	my $nr = 1345345333;
	my $add = 7;
	my $nr2 = 0x12345671;
	my $tmp;
	my $pwlen = length $password;
	my $c;

	for (my $i=0; $i < $pwlen; $i++) {
		my $c = substr $password, $i, 1;
		next if $c eq ' ' || $c eq "\t";
		my $tmp = ord $c;
		my $value = ((_and_by_char($nr, 63) + $add) * $tmp) + $nr * 256;
		$nr = _xor_by_long($nr, $value);
		$nr2 += _xor_by_long(($nr2 * 256), $nr);
		$add += $tmp;
	}
	return (_and_by_long($nr, 0x7fffffff), _and_by_long($nr2, 0x7fffffff));
}


sub _and_by_char
{
	my $source = shift;
	my $mask   = shift;

	return $source & $mask;
}


sub _and_by_long
{
	my $source = shift;
	my $mask = shift || 0xFFFFFFFF;

	return _cut_off_to_long($source) & _cut_off_to_long($mask);
}


sub _xor_by_long
{
	my $source = shift;
	my $mask = shift || 0;

	return _cut_off_to_long($source) ^ _cut_off_to_long($mask);
}


sub _cut_off_to_long
{
	my $source = shift;

	if ($] >= 5.006) {
		$source = $source % (0xFFFFFFFF + 1) if $source > 0xFFFFFFFF;
		return $source;
	}
	while ($source > 0xFFFFFFFF) {
		$source -= 0xFFFFFFFF + 1;
	}
	return $source;
}


1;
__END__