File Coverage

blib/lib/Mojo/SMTP/Client.pm

Criterion	Covered	Total	%
statement	301	314	95.8
branch	74	96	77.0
condition	10	14	71.4
subroutine	60	62	96.7
pod	3	3	100.0
total	448	489	91.6

package Mojo::SMTP::Client;

use Mojo::Base 'Mojo::EventEmitter';
use Mojo::IOLoop;
use Mojo::IOLoop::Client;
use Mojo::IOLoop::Delay;
use Mojo::IOLoop::Stream;
use Mojo::Util 'b64_encode';
use Mojo::SMTP::Client::Response;
use Mojo::SMTP::Client::Exception;
use Scalar::Util 'weaken';
use Carp;

our $VERSION = '0.19';

use constant {
	CMD_OK       => 2,
	CMD_MORE     => 3,

	CMD_CONNECT  => 1,
	CMD_EHLO     => 2,
	CMD_HELO     => 3,
	CMD_STARTTLS => 10,
	CMD_AUTH     => 11,
	CMD_FROM     => 4,
	CMD_TO       => 5,
	CMD_DATA     => 6,
	CMD_DATA_END => 7,
	CMD_RESET    => 8,
	CMD_QUIT     => 9,
};

our %CMD = (
	&CMD_CONNECT  => 'CMD_CONNECT',
	&CMD_EHLO     => 'CMD_EHLO',
	&CMD_HELO     => 'CMD_HELO',
	&CMD_STARTTLS => 'CMD_STARTTLS',
	&CMD_AUTH     => 'CMD_AUTH',
	&CMD_FROM     => 'CMD_FROM',
	&CMD_TO       => 'CMD_TO',
	&CMD_DATA     => 'CMD_DATA',
	&CMD_DATA_END => 'CMD_DATA_END',
	&CMD_RESET    => 'CMD_RESET',
	&CMD_QUIT     => 'CMD_QUIT',
);

has address            => 'localhost';
has port               => sub { $_[0]->tls ? 465 : 25 };
has tls                => 0;
has 'tls_ca';
has 'tls_cert';
has 'tls_key';
has tls_verify         => 1;
has hello              => 'localhost.localdomain';
has connect_timeout    => sub { $ENV{MOJO_CONNECT_TIMEOUT} || 10 };
has inactivity_timeout => sub { $ENV{MOJO_INACTIVITY_TIMEOUT} // 20 };
has ioloop             => sub { Mojo::IOLoop->new };
has autodie            => 0;

sub new {
	my $class = shift;

	my $self = $class->SUPER::new(@_);
	weaken(my $this = $self);

	$self->{resp_checker} = sub {
		my ($delay, $resp) = @_;
		$this->emit(response => $this->{last_cmd}, $resp);

		unless (substr($resp->code, 0, 1) == $this->{expected_code}) {
			die $resp->error(Mojo::SMTP::Client::Exception::Response->new($resp->message)->code($resp->code));
		}
		$delay->pass($resp);
	};

	$self->{cmds} = [];

	$self;
}

sub send {
	my $self = shift;
	my $cb = @_ % 2 == 0 ? undef : pop;

	my @steps;
	$self->{nb} = $cb ? 1 : 0;

	# user changed SMTP server or server sent smth while it shouldn't
	if ($self->{stream} && (($self->{server} ne $self->_server) ||
	     ($self->{stream}->is_readable && !$self->{starttls} && !$self->{authorized} &&
	      grep {$self->{last_cmd} == $_} (CMD_CONNECT, CMD_DATA_END, CMD_RESET)))
	) {
		$self->_rm_stream();
	}

	unless ($self->{stream}) {
		push @steps, sub {
			my $delay = shift;
			# connect
			$self->{starttls} = $self->{authorized} = 0;
			$self->emit('start');
			$self->{server} = $self->_server;
			$self->{last_cmd} = CMD_CONNECT;

			my $connect_cb = $delay->begin;
			$self->{client} = Mojo::IOLoop::Client->new(reactor => $self->_ioloop->reactor);
			$self->{client}->on(connect => $connect_cb);
			$self->{client}->on(error => $connect_cb);
			$self->{client}->connect(
				address    => $self->address,
				port       => $self->port,
				timeout    => $self->connect_timeout,
				tls        => $self->tls,
				tls_ca     => $self->tls_ca,
				tls_cert   => $self->tls_cert,
				tls_key    => $self->tls_key,
				tls_options => { SSL_verify_mode => $self->tls_verify },
			);
		},
		sub {
			# read response
			my $delay = shift;
			delete $self->{client};
			# check is this a handle
			Mojo::SMTP::Client::Exception::Stream->throw($_[0]) unless eval { *{$_[0]} };

			$self->_make_stream($_[0], $self->_ioloop);
			$self->_read_response($delay->begin);
			$self->{expected_code} = CMD_OK;
		},
		# check response
		$self->{resp_checker};

		if (!@_ || $_[0] ne 'hello') {
			unshift @_, hello => $self->hello;
		}
	}
	else {
		$self->{stream}->start;
	}

	push @{$self->{cmds}}, @_;
	push @steps, $self->_make_cmd_steps();

	# non-blocking
	my $delay = $self->{delay} = Mojo::IOLoop::Delay->new->ioloop($self->_ioloop)->steps(@steps);
	$self->{finally} = sub {
		shift if @_ == 2; # delay

		if ($cb) {
			my $r = $_[0];
			unless ($r->isa('Mojo::SMTP::Client::Response')) {
				# some error occured, which throwed an exception
				$r = Mojo::SMTP::Client::Response->new('', error => $r);
			}

			delete $self->{delay};
			delete $self->{finally};

			$cb->($self, $r);
			$cb = undef;
		}
	};
	$delay->catch($self->{finally});

	# blocking
	my $resp;
	unless ($self->{nb}) {
		$cb = sub {
			$resp = pop;
		};
		$delay->wait;
		return $self->autodie && $resp->error ? die $resp->error : $resp;
	}
}

sub prepend_cmd {
	my $self = shift;
	croak "no active `send' calls" unless exists $self->{delay};

	unshift @{ $self->{cmds} }, @_;
}

sub _ioloop {
	my ($self) = @_;
	return $self->{nb} ? Mojo::IOLoop->singleton : $self->ioloop;
}

sub _server {
	my $self = shift;
	return $self->address.':'.$self->port.':'.$self->tls;
}

sub _make_stream {
	my ($self, $sock, $loop) = @_;

	weaken $self;
	my $error_handler = sub {
		delete($self->{cleanup_cb})->() if $self->{cleanup_cb};
		$self->_rm_stream();

		$self->{delay}->reject($_[0]);
	};

	$self->{stream} = Mojo::IOLoop::Stream->new($sock);
	$self->{stream}->reactor($loop->reactor);
	$self->{stream}->start;
	$self->{stream}->on(timeout => sub {
		$error_handler->(Mojo::SMTP::Client::Exception::Stream->new('Inactivity timeout'));
	});
	$self->{stream}->on(error => sub {
		$error_handler->(Mojo::SMTP::Client::Exception::Stream->new($_[-1]));
	});
	$self->{stream}->on(close => sub {
		$error_handler->(Mojo::SMTP::Client::Exception::Stream->new('Socket closed unexpectedly by remote side'));
	});
}

sub _make_cmd_steps {
	my ($self) = @_;

	my ($cmd, $arg) = splice @{ $self->{cmds} }, 0, 2;
	unless ($cmd) {
		# no more commands
		if ($self->{stream}) {
			$self->{stream}->timeout(0);
			$self->{stream}->stop;
		}
		return $self->{finally};
	}

	if ( my $sub = $self->can("_cmd_$cmd") ) {
		return (
			$self->$sub($arg), sub {
				my ($delay, $resp) = @_;

				$delay->pass($resp);
				$delay->steps( $self->_make_cmd_steps() );
			}
		);
	}

	croak 'unrecognized command: ', $cmd;
}

# EHLO/HELO
sub _cmd_hello {
	my ($self, $arg) = @_;
	weaken $self;

	return (
		sub {
			my $delay = shift;
			$self->_write_cmd('EHLO ' . $arg, CMD_EHLO);
			$self->_read_response($delay->begin);
			$self->{expected_code} = CMD_OK;
		},
		sub {
			eval { $self->{resp_checker}->(@_); $_[1]->{checked} = 1 };
			if (my $e = $@) {
				die $e unless $e->isa('Mojo::SMTP::Client::Response');
				my $delay = shift;

				$self->_write_cmd('HELO ' . $arg, CMD_HELO);
				$self->_read_response($delay->begin);
			}
		},
		sub {
			my ($delay, $resp) = @_;
			return $delay->pass($resp) if delete $resp->{checked};
			$self->{resp_checker}->($delay, $resp);
		}
	);
}

# STARTTLS
sub _cmd_starttls {
	my ($self, $arg) = @_;
	weaken $self;

	require IO::Socket::SSL and IO::Socket::SSL->VERSION(0.98);

	return (
		sub {
			my $delay = shift;
			$self->_write_cmd('STARTTLS', CMD_STARTTLS);
			$self->_read_response($delay->begin);
			$self->{expected_code} = CMD_OK;
		},
		$self->{resp_checker},
		sub {
			my ($delay, $resp) = @_;
			$self->{stream}->stop;
			$self->{stream}->timeout(0);

			my ($tls_cb, $tid, $loop, $sock);

			my $error_handler = sub {
				$loop->remove($tid);
				$loop->reactor->remove($sock);
				$sock = undef;
				$tls_cb->($delay, undef, @_>=2 ? $_[1] : 'Inactivity timeout');
				$tls_cb = $delay = undef;
			};

			$sock = IO::Socket::SSL->start_SSL(
				$self->{stream}->steal_handle,
				SSL_ca_file         => $self->tls_ca,
				SSL_cert_file       => $self->tls_cert,
				SSL_key_file        => $self->tls_key,
				SSL_verify_mode     => $self->tls_verify,
				SSL_verifycn_name   => $self->address,
				SSL_verifycn_scheme => $self->tls_ca ? 'smtp' : undef,
				SSL_startHandshake  => 0,
				SSL_error_trap      => $error_handler
			)
			or return $delay->pass(0, $IO::Socket::SSL::SSL_ERROR);

			$tls_cb = $delay->begin;
			$loop = $self->_ioloop;

			$tid = $loop->timer($self->inactivity_timeout => $error_handler);

			$loop->reactor->io($sock => sub {
				if ($sock->connect_SSL) {
					$loop->remove($tid);
					$loop->reactor->remove($sock);
					$self->_make_stream($sock, $loop);
					$self->{starttls} = 1;
					$sock = $loop = undef;
					$tls_cb->($delay, $resp);
					$tls_cb = $delay = undef;
					return;
				}

				return $loop->reactor->watch($sock, 1, 0)
					if $IO::Socket::SSL::SSL_ERROR == IO::Socket::SSL::SSL_WANT_READ();
				return $loop->reactor->watch($sock, 0, 1)
					if $IO::Socket::SSL::SSL_ERROR == IO::Socket::SSL::SSL_WANT_WRITE();

			})->watch($sock, 0, 1);
		},
		sub {
			my ($delay, $resp, $error) = @_;
			unless ($resp) {
				$self->_rm_stream();
				Mojo::SMTP::Client::Exception::Stream->throw($error);
			}

			$delay->pass($resp);
		}
	);
}

# AUTH
sub _cmd_auth {
	my ($self, $arg) = @_;
	weaken $self;

	my $type = lc($arg->{type} // 'plain');

	my $set_auth_ok = sub {
		my ($delay, $resp) = @_;
		$self->{authorized} = 1;
		$delay->pass($resp);
	};

	if ($type eq 'plain') {
		return (
			sub {
				my $delay = shift;
				$self->_write_cmd('AUTH PLAIN '.b64_encode(join("\0", '', $arg->{login}, $arg->{password}), ''), CMD_AUTH);
				$self->_read_response($delay->begin);
				$self->{expected_code} = CMD_OK;
			},
			$self->{resp_checker},
			$set_auth_ok
		);
	}

	if ($type eq 'login') {
		return (
			# start auth
			sub {
				my $delay = shift;
				$self->_write_cmd('AUTH LOGIN', CMD_AUTH);
				$self->_read_response($delay->begin);
				$self->{expected_code} = CMD_MORE;
			},
			$self->{resp_checker},
			# send username
			sub {
				my $delay = shift;
				$self->_write_cmd(b64_encode($arg->{login}, ''), CMD_AUTH);
				$self->_read_response($delay->begin);
				$self->{expected_code} = CMD_MORE;
			},
			$self->{resp_checker},
			# send password
			sub {
				my $delay = shift;
				$self->_write_cmd(b64_encode($arg->{password}, ''), CMD_AUTH);
				$self->_read_response($delay->begin);
				$self->{expected_code} = CMD_OK;
			},
			$self->{resp_checker},
			$set_auth_ok
		);
	}

	croak 'unrecognized auth method: ', $type;
}

# FROM
sub _cmd_from {
	my ($self, $arg) = @_;
	weaken $self;

	return (
		sub {
			my $delay = shift;
			$self->_write_cmd('MAIL FROM:<'.$arg.'>', CMD_FROM);
			$self->_read_response($delay->begin);
			$self->{expected_code} = CMD_OK;
		},
		$self->{resp_checker}
	);
}

# TO
sub _cmd_to {
	my ($self, $arg) = @_;
	weaken $self;

	my @steps;

	for my $to (ref $arg ? @$arg : $arg) {
		push @steps, sub {
			my $delay = shift;
			$self->_write_cmd('RCPT TO:<'.$to.'>', CMD_TO);
			$self->_read_response($delay->begin);
			$self->{expected_code} = CMD_OK;
		},
		$self->{resp_checker}
	}

	return @steps;
}

# DATA
sub _cmd_data {
	my ($self, $arg) = @_;
	weaken $self;

	my @steps;

	push @steps, sub {
		my $delay = shift;
		$self->_write_cmd('DATA', CMD_DATA);
		$self->_read_response($delay->begin);
		$self->{expected_code} = CMD_MORE;
	},
	$self->{resp_checker};

	if (ref $arg eq 'CODE') {
		my ($data_writer, $data_writer_cb);
		my $was_nl;
		my $last_ch;

		$data_writer = sub {
			my $delay = shift;
			unless ($data_writer_cb) {
				$data_writer_cb = $delay->begin;
				$self->{cleanup_cb} = sub {
					undef $data_writer;
				};
			}

			my $data = $arg->();
			$data = $$data if ref $data;

			unless (length($data) > 0) {
				$self->_write_cmd(($was_nl ? '' : Mojo::SMTP::Client::Response::CRLF).'.', CMD_DATA_END);
				$self->_read_response($data_writer_cb);
				$self->{expected_code} = CMD_OK;
				return delete($self->{cleanup_cb})->();
			}
			# The following part if heavily inspired by Net::Cmd
			my $first_ch = '';
			# We have not send anything yet, so last_ch = "\012" means we are at the start of a line (^. -> ..)
			$last_ch = "\012" unless defined $last_ch;
			if ($last_ch eq "\015") {
				# Remove \012 so it does not get prefixed with another \015 below
				# and escape the . if there is one following it because the fixup
				# below will not find it
				$first_ch = "\012" if $data =~ s/^\012(\.?)/$1$1/;
			}
			elsif ($last_ch eq "\012") {
				# Fixup below will not find the . as the first character of the buffer
				$first_ch = "." if $data =~ /^\./;
			}
			$data =~ s/\015?\012(\.?)/\015\012$1$1/g;
			substr($data, 0, 0) = $first_ch;
			$last_ch = substr($data, -1, 1);
			$was_nl = _has_nl($data);
			$self->{stream}->write($data, $data_writer);
		};

		push @steps, $data_writer, $self->{resp_checker};
	}
	else {
		push @steps, sub {
			my $delay = shift;
			(ref $arg ? $$arg : $arg) =~ s/\015?\012(\.?)/\015\012$1$1/g; # turn . into .. if it's first character of the line and normalize newline
			$self->{stream}->write(ref $arg ? $$arg : $arg, $delay->begin);
		},
		sub {
			my $delay = shift;
			$self->_write_cmd((_has_nl($arg) ? '' : Mojo::SMTP::Client::Response::CRLF).'.', CMD_DATA_END);
			$self->_read_response($delay->begin);
			$self->{expected_code} = CMD_OK;
		},
		$self->{resp_checker}
	}

	return @steps;
}

# RESET
sub _cmd_reset {
	my ($self, $arg) = @_;
	weaken $self;

	return (
		sub {
			my $delay = shift;
			$self->_write_cmd('RSET', CMD_RESET);
			$self->_read_response($delay->begin);
			$self->{expected_code} = CMD_OK;
		},
		$self->{resp_checker}
	);
}

# QUIT
sub _cmd_quit {
	my ($self, $arg) = @_;
	weaken $self;

	return (
		sub {
			my $delay = shift;
			$self->_write_cmd('QUIT', CMD_QUIT);
			$self->_read_response($delay->begin);
			$self->{expected_code} = CMD_OK;
		},
		$self->{resp_checker}, sub {
			my $delay = shift;
			$self->_rm_stream();
			$delay->pass(@_);
		}
	);
}

sub _write_cmd {
	my ($self, $cmd, $cmd_const) = @_;
	$self->{last_cmd} = $cmd_const;
	$self->{stream}->write($cmd.Mojo::SMTP::Client::Response::CRLF);
}

sub _read_response {
	my ($self, $cb) = @_;
	$self->{stream}->timeout($self->inactivity_timeout);
	my $resp = '';

	$self->{stream}->on(read => sub {
		$resp .= $_[-1];
		if ($resp =~ /^\d+(?:\s[^\n]*)?\n$/m) {
			$self->{stream}->unsubscribe('read');
			$cb->($self, Mojo::SMTP::Client::Response->new($resp));
		}
	});
}

sub _rm_stream {
	my $self = shift;
	$self->{stream}->unsubscribe('close')
	               ->unsubscribe('timeout')
	               ->unsubscribe('error')
	               ->unsubscribe('read');
	delete $self->{stream};
}

sub _has_nl {
	substr(ref $_[0] ? ${$_[0]} : $_[0], -2, 2) eq Mojo::SMTP::Client::Response::CRLF;
}

sub DESTROY {
	my $self = shift;
	if ($self->{stream}) {
		$self->_rm_stream();
	}
}

1;

__END__