Безопасность, SMTP, очередь отправки, DKIM, деплой на Jambotron (Jambotron v0.2.3 75.1–75.3)
- SMTP-сервер переписан: разбор по строкам, AUTH PLAIN/LOGIN, нет открытого релея, MAIL FROM только свой после входа, 250 только после сохранения (иначе 451), лимит размера - отправка наружу: очередь outbound_queue (V2) с повторами ~22 ч и уведомлением о недоставке; Jakarta Mail, режимы off / mx / smarthost, байты письма не меняются (DKIM) - DKIM: подпись utils-mail-dkim, проверка Apache jDKIM; ключ PKCS#8, старый PKCS#1 читается - права: письма, папки, вложения — только владельцу (было: любой по номеру); имена вложений без ../ - секреты без значений по умолчанию (DB_PASSWORD, JWT_SECRET), регистрация закрыта, admin123 и /api/simulate/receive — только профиль dev - веб-клиент внутри jar (относительный /api), Dockerfile в 3 этапа, не root, health; docker-compose.jambotron.yml: секреты создаются при первом запуске, наружу только 8080 - gradle-wrapper.jar был в .gitignore — сборка не запускалась - тесты: 26 (SMTP реальным клиентом и сокетом, GreenMail, DKIM подпись+проверка, очередь) α Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
jambotron α
@@ -0,0 +1,96 @@
|
||||
package com.mailserver.delivery;
|
||||
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import org.springframework.scheduling.annotation.Scheduled;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
import java.time.Duration;
|
||||
import java.time.Instant;
|
||||
import java.util.List;
|
||||
|
||||
/**
|
||||
* Sends queued messages. Temporary failure (4xx, no connection, DNS) — retry later with growing pauses,
|
||||
* about 22 hours in total; permanent failure (5xx) or retries used up — FAILED and a bounce to the sender's inbox.
|
||||
* Runs on the scheduler thread: blocking here is fine, it is not the network event loop.
|
||||
*/
|
||||
@Component
|
||||
public class DeliveryWorker {
|
||||
|
||||
private static final Logger log = LoggerFactory.getLogger(DeliveryWorker.class);
|
||||
|
||||
static final List<Duration> RETRY_PAUSES = List.of(
|
||||
Duration.ofMinutes(1), Duration.ofMinutes(5), Duration.ofMinutes(15), Duration.ofHours(1),
|
||||
Duration.ofHours(3), Duration.ofHours(6), Duration.ofHours(12));
|
||||
private static final int BATCH = 20;
|
||||
|
||||
/** Tells the sender that a message did not reach someone. */
|
||||
public interface BounceSender {
|
||||
void bounce(OutboundMessage message, String reason);
|
||||
}
|
||||
|
||||
private final OutboundRepository repository;
|
||||
private final SmtpDeliveryClient client;
|
||||
private final BounceSender bounces;
|
||||
|
||||
public DeliveryWorker(OutboundRepository repository, SmtpDeliveryClient client, BounceSender bounces) {
|
||||
this.repository = repository;
|
||||
this.client = client;
|
||||
this.bounces = bounces;
|
||||
}
|
||||
|
||||
@Scheduled(initialDelayString = "${mailserver.queue.initial-delay-ms:5000}", fixedDelayString = "${mailserver.queue.poll-ms:15000}")
|
||||
public void deliverDue() {
|
||||
List<OutboundMessage> due = repository.findDue(Instant.now(), BATCH).collectList().block();
|
||||
if (due == null) {
|
||||
return;
|
||||
}
|
||||
for (OutboundMessage message : due) {
|
||||
try {
|
||||
attempt(message, Instant.now());
|
||||
} catch (RuntimeException e) {
|
||||
log.error("Queue item {} not processed: {}", message.getId(), e.toString());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
void attempt(OutboundMessage message, Instant now) {
|
||||
SmtpDeliveryClient.Result result = client.deliver(message.getMailFrom(), message.getRcptTo(), message.getMessage());
|
||||
message.setAttempts(message.getAttempts() + 1);
|
||||
message.setLastError(result.outcome() == SmtpDeliveryClient.Outcome.DELIVERED ? null : truncate(result.detail()));
|
||||
switch (result.outcome()) {
|
||||
case DELIVERED -> finish(message, OutboundMessage.DELIVERED, now);
|
||||
case PERMANENT -> fail(message, now, result.detail());
|
||||
case TEMPORARY -> {
|
||||
if (message.getAttempts() > RETRY_PAUSES.size()) {
|
||||
fail(message, now, "gave up after " + message.getAttempts() + " attempts: " + result.detail());
|
||||
} else {
|
||||
message.setNextAttemptAt(now.plus(RETRY_PAUSES.get(message.getAttempts() - 1)));
|
||||
repository.save(message).block();
|
||||
log.info("Queue item {} to {}: retry at {} ({})", message.getId(), message.getRcptTo(),
|
||||
message.getNextAttemptAt(), result.detail());
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private void fail(OutboundMessage message, Instant now, String reason) {
|
||||
finish(message, OutboundMessage.FAILED, now);
|
||||
log.warn("Queue item {} to {} FAILED: {}", message.getId(), message.getRcptTo(), reason);
|
||||
try {
|
||||
bounces.bounce(message, reason);
|
||||
} catch (RuntimeException e) {
|
||||
log.error("Bounce for queue item {} not stored: {}", message.getId(), e.toString());
|
||||
}
|
||||
}
|
||||
|
||||
private void finish(OutboundMessage message, String status, Instant now) {
|
||||
message.setStatus(status);
|
||||
message.setFinishedAt(now);
|
||||
repository.save(message).block();
|
||||
}
|
||||
|
||||
private static String truncate(String text) {
|
||||
return text == null || text.length() <= 1000 ? text : text.substring(0, 1000);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,57 @@
|
||||
package com.mailserver.delivery;
|
||||
|
||||
import org.springframework.data.annotation.Id;
|
||||
import org.springframework.data.relational.core.mapping.Column;
|
||||
import org.springframework.data.relational.core.mapping.Table;
|
||||
|
||||
import java.time.Instant;
|
||||
|
||||
@Table("outbound_queue")
|
||||
public class OutboundMessage {
|
||||
|
||||
public static final String PENDING = "PENDING";
|
||||
public static final String DELIVERED = "DELIVERED";
|
||||
public static final String FAILED = "FAILED";
|
||||
|
||||
@Id
|
||||
private Long id;
|
||||
@Column("mail_from")
|
||||
private String mailFrom;
|
||||
@Column("rcpt_to")
|
||||
private String rcptTo;
|
||||
@Column("message")
|
||||
private byte[] message;
|
||||
@Column("status")
|
||||
private String status = PENDING;
|
||||
@Column("attempts")
|
||||
private int attempts;
|
||||
@Column("next_attempt_at")
|
||||
private Instant nextAttemptAt;
|
||||
@Column("last_error")
|
||||
private String lastError;
|
||||
@Column("created_at")
|
||||
private Instant createdAt;
|
||||
@Column("finished_at")
|
||||
private Instant finishedAt;
|
||||
|
||||
public Long getId() { return id; }
|
||||
public void setId(Long id) { this.id = id; }
|
||||
public String getMailFrom() { return mailFrom; }
|
||||
public void setMailFrom(String mailFrom) { this.mailFrom = mailFrom; }
|
||||
public String getRcptTo() { return rcptTo; }
|
||||
public void setRcptTo(String rcptTo) { this.rcptTo = rcptTo; }
|
||||
public byte[] getMessage() { return message; }
|
||||
public void setMessage(byte[] message) { this.message = message; }
|
||||
public String getStatus() { return status; }
|
||||
public void setStatus(String status) { this.status = status; }
|
||||
public int getAttempts() { return attempts; }
|
||||
public void setAttempts(int attempts) { this.attempts = attempts; }
|
||||
public Instant getNextAttemptAt() { return nextAttemptAt; }
|
||||
public void setNextAttemptAt(Instant nextAttemptAt) { this.nextAttemptAt = nextAttemptAt; }
|
||||
public String getLastError() { return lastError; }
|
||||
public void setLastError(String lastError) { this.lastError = lastError; }
|
||||
public Instant getCreatedAt() { return createdAt; }
|
||||
public void setCreatedAt(Instant createdAt) { this.createdAt = createdAt; }
|
||||
public Instant getFinishedAt() { return finishedAt; }
|
||||
public void setFinishedAt(Instant finishedAt) { this.finishedAt = finishedAt; }
|
||||
}
|
||||
@@ -0,0 +1,28 @@
|
||||
package com.mailserver.delivery;
|
||||
|
||||
import org.springframework.stereotype.Service;
|
||||
import reactor.core.publisher.Mono;
|
||||
|
||||
import java.time.Instant;
|
||||
|
||||
/** Puts a message for one external recipient into the delivery queue (the worker sends it). */
|
||||
@Service
|
||||
public class OutboundQueue {
|
||||
|
||||
private final OutboundRepository repository;
|
||||
|
||||
public OutboundQueue(OutboundRepository repository) {
|
||||
this.repository = repository;
|
||||
}
|
||||
|
||||
public Mono<OutboundMessage> enqueue(String mailFrom, String rcptTo, byte[] signedMessage) {
|
||||
OutboundMessage message = new OutboundMessage();
|
||||
message.setMailFrom(mailFrom);
|
||||
message.setRcptTo(rcptTo);
|
||||
message.setMessage(signedMessage);
|
||||
message.setStatus(OutboundMessage.PENDING);
|
||||
message.setCreatedAt(Instant.now());
|
||||
message.setNextAttemptAt(Instant.now());
|
||||
return repository.save(message);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,13 @@
|
||||
package com.mailserver.delivery;
|
||||
|
||||
import org.springframework.data.r2dbc.repository.Query;
|
||||
import org.springframework.data.r2dbc.repository.R2dbcRepository;
|
||||
import reactor.core.publisher.Flux;
|
||||
|
||||
import java.time.Instant;
|
||||
|
||||
public interface OutboundRepository extends R2dbcRepository<OutboundMessage, Long> {
|
||||
|
||||
@Query("SELECT * FROM outbound_queue WHERE status = 'PENDING' AND next_attempt_at <= :now ORDER BY next_attempt_at LIMIT :limit")
|
||||
Flux<OutboundMessage> findDue(Instant now, int limit);
|
||||
}
|
||||
@@ -0,0 +1,244 @@
|
||||
package com.mailserver.delivery;
|
||||
|
||||
import jakarta.mail.Address;
|
||||
import jakarta.mail.AuthenticationFailedException;
|
||||
import jakarta.mail.MessagingException;
|
||||
import jakarta.mail.SendFailedException;
|
||||
import jakarta.mail.Session;
|
||||
import jakarta.mail.Transport;
|
||||
import jakarta.mail.internet.InternetAddress;
|
||||
import jakarta.mail.internet.MimeMessage;
|
||||
import org.eclipse.angus.mail.smtp.SMTPAddressFailedException;
|
||||
import org.eclipse.angus.mail.smtp.SMTPSendFailedException;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import org.xbill.DNS.Lookup;
|
||||
import org.xbill.DNS.MXRecord;
|
||||
import org.xbill.DNS.Record;
|
||||
import org.xbill.DNS.TextParseException;
|
||||
import org.xbill.DNS.Type;
|
||||
|
||||
import java.io.ByteArrayInputStream;
|
||||
import java.io.IOException;
|
||||
import java.io.OutputStream;
|
||||
import java.util.Arrays;
|
||||
import java.util.Comparator;
|
||||
import java.util.List;
|
||||
import java.util.Locale;
|
||||
import java.util.Properties;
|
||||
|
||||
/**
|
||||
* Delivers one message to one recipient over SMTP (Jakarta Mail), blocking.
|
||||
* <ul>
|
||||
* <li><b>mx</b> — straight to the recipient's MX servers on port 25, lowest preference first; STARTTLS when offered,
|
||||
* without certificate checks — opportunistic, like Postfix {@code smtp_tls_security_level=may};</li>
|
||||
* <li><b>smarthost</b> — everything through one provider (Resend, SES, …) with login and verified TLS;</li>
|
||||
* <li><b>off</b> — nothing leaves the server.</li>
|
||||
* </ul>
|
||||
* The message bytes are sent exactly as given (they carry the DKIM signature).
|
||||
*/
|
||||
public class SmtpDeliveryClient {
|
||||
|
||||
private static final Logger log = LoggerFactory.getLogger(SmtpDeliveryClient.class);
|
||||
|
||||
public enum Mode { OFF, MX, SMARTHOST }
|
||||
|
||||
public enum Tls { NONE, STARTTLS, SSL }
|
||||
|
||||
/** port — of the smarthost; mxPort — of recipients' MX servers (25; another value only in tests). */
|
||||
public record Settings(Mode mode, String host, int port, String username, String password, Tls tls,
|
||||
String heloName, int timeoutMs, int mxPort) {
|
||||
}
|
||||
|
||||
public enum Outcome { DELIVERED, TEMPORARY, PERMANENT }
|
||||
|
||||
public record Result(Outcome outcome, String detail) {
|
||||
static Result delivered(String detail) { return new Result(Outcome.DELIVERED, detail); }
|
||||
static Result temporary(String detail) { return new Result(Outcome.TEMPORARY, detail); }
|
||||
static Result permanent(String detail) { return new Result(Outcome.PERMANENT, detail); }
|
||||
}
|
||||
|
||||
/** MX hosts of a domain, best first; empty — the domain has no mail server. */
|
||||
public interface MxResolver {
|
||||
List<String> resolve(String domain) throws TemporaryDnsException;
|
||||
}
|
||||
|
||||
public static class TemporaryDnsException extends Exception {
|
||||
public TemporaryDnsException(String message) {
|
||||
super(message);
|
||||
}
|
||||
}
|
||||
|
||||
private final Settings settings;
|
||||
private final MxResolver resolver;
|
||||
|
||||
public SmtpDeliveryClient(Settings settings) {
|
||||
this(settings, SmtpDeliveryClient::dnsMx);
|
||||
}
|
||||
|
||||
public SmtpDeliveryClient(Settings settings, MxResolver resolver) {
|
||||
this.settings = settings;
|
||||
this.resolver = resolver;
|
||||
}
|
||||
|
||||
public Mode mode() {
|
||||
return settings.mode();
|
||||
}
|
||||
|
||||
public Result deliver(String from, String recipient, byte[] message) {
|
||||
return switch (settings.mode()) {
|
||||
case OFF -> Result.permanent("outbound delivery is disabled on this server");
|
||||
case SMARTHOST -> send(settings.host(), settings.port(), true, from, recipient, message);
|
||||
case MX -> viaMx(from, recipient, message);
|
||||
};
|
||||
}
|
||||
|
||||
private Result viaMx(String from, String recipient, byte[] message) {
|
||||
String domain = recipient.substring(recipient.lastIndexOf('@') + 1).toLowerCase(Locale.ROOT);
|
||||
List<String> hosts;
|
||||
try {
|
||||
hosts = resolver.resolve(domain);
|
||||
} catch (TemporaryDnsException e) {
|
||||
return Result.temporary("DNS: " + e.getMessage());
|
||||
}
|
||||
if (hosts.isEmpty()) {
|
||||
return Result.permanent("domain " + domain + " does not accept mail (no MX/A record)");
|
||||
}
|
||||
Result last = null;
|
||||
for (String host : hosts) {
|
||||
last = send(host, settings.mxPort(), false, from, recipient, message);
|
||||
// A server answered with a decision — do not ask the next MX; only a connection problem moves on
|
||||
if (last.outcome() != Outcome.TEMPORARY || last.detail().startsWith("4")) {
|
||||
return last;
|
||||
}
|
||||
}
|
||||
return last;
|
||||
}
|
||||
|
||||
private Result send(String host, int port, boolean smarthost, String from, String recipient, byte[] message) {
|
||||
Properties props = new Properties();
|
||||
props.put("mail.smtp.host", host);
|
||||
props.put("mail.smtp.port", String.valueOf(port));
|
||||
props.put("mail.smtp.localhost", settings.heloName());
|
||||
props.put("mail.smtp.connectiontimeout", String.valueOf(settings.timeoutMs()));
|
||||
props.put("mail.smtp.timeout", String.valueOf(settings.timeoutMs()));
|
||||
props.put("mail.smtp.writetimeout", String.valueOf(settings.timeoutMs()));
|
||||
props.put("mail.smtp.from", from);
|
||||
props.put("mail.smtp.allow8bitmime", "true");
|
||||
boolean login = false;
|
||||
if (smarthost) {
|
||||
switch (settings.tls()) {
|
||||
case STARTTLS -> {
|
||||
props.put("mail.smtp.starttls.enable", "true");
|
||||
props.put("mail.smtp.starttls.required", "true");
|
||||
}
|
||||
case SSL -> props.put("mail.smtp.ssl.enable", "true");
|
||||
case NONE -> { }
|
||||
}
|
||||
props.put("mail.smtp.ssl.checkserveridentity", "true");
|
||||
login = settings.username() != null && !settings.username().isBlank();
|
||||
props.put("mail.smtp.auth", String.valueOf(login));
|
||||
} else {
|
||||
props.put("mail.smtp.starttls.enable", "true");
|
||||
props.put("mail.smtp.ssl.trust", "*");
|
||||
}
|
||||
Session session = Session.getInstance(props);
|
||||
try (Transport transport = session.getTransport(settings.tls() == Tls.SSL && smarthost ? "smtps" : "smtp")) {
|
||||
if (login) {
|
||||
transport.connect(host, port, settings.username(), settings.password());
|
||||
} else {
|
||||
transport.connect(host, port, null, null);
|
||||
}
|
||||
transport.sendMessage(new RawMessage(session, message), new Address[]{new InternetAddress(recipient)});
|
||||
log.info("Delivered {} -> {} via {}:{}", from, recipient, host, port);
|
||||
return Result.delivered(host);
|
||||
} catch (AuthenticationFailedException e) {
|
||||
return Result.temporary("login to " + host + " failed: " + e.getMessage());
|
||||
} catch (MessagingException e) {
|
||||
int code = replyCode(e);
|
||||
String text = code > 0 ? code + " " + oneLine(e) : "connection to " + host + ": " + oneLine(e);
|
||||
log.warn("Delivery {} -> {} via {}: {}", from, recipient, host, text);
|
||||
return code >= 500 ? Result.permanent(text) : Result.temporary(text);
|
||||
}
|
||||
}
|
||||
|
||||
/** SMTP reply code from the exception chain, 0 — no reply (connection problem). */
|
||||
static int replyCode(Exception e) {
|
||||
Exception current = e;
|
||||
while (current != null) {
|
||||
if (current instanceof SMTPAddressFailedException f) {
|
||||
return f.getReturnCode();
|
||||
}
|
||||
if (current instanceof SMTPSendFailedException f) {
|
||||
return f.getReturnCode();
|
||||
}
|
||||
current = current instanceof MessagingException m ? m.getNextException()
|
||||
: current.getCause() instanceof Exception c ? c : null;
|
||||
if (current instanceof SendFailedException sf && sf.getInvalidAddresses() != null
|
||||
&& sf.getNextException() == null) {
|
||||
return 550;
|
||||
}
|
||||
}
|
||||
return 0;
|
||||
}
|
||||
|
||||
private static String oneLine(Exception e) {
|
||||
String message = String.valueOf(e.getMessage()).replaceAll("\\s+", " ").trim();
|
||||
return message.length() > 500 ? message.substring(0, 500) : message;
|
||||
}
|
||||
|
||||
/** MX hosts by preference (RFC 5321 5.1); no MX — the domain itself (implicit MX); null MX "." — none. */
|
||||
static List<String> dnsMx(String domain) throws TemporaryDnsException {
|
||||
try {
|
||||
Lookup lookup = new Lookup(domain, Type.MX);
|
||||
Record[] records = lookup.run();
|
||||
switch (lookup.getResult()) {
|
||||
case Lookup.SUCCESSFUL -> {
|
||||
List<String> hosts = Arrays.stream(records)
|
||||
.map(MXRecord.class::cast)
|
||||
.sorted(Comparator.comparingInt(MXRecord::getPriority))
|
||||
.map(mx -> mx.getTarget().toString(true))
|
||||
.filter(host -> !host.isEmpty() && !host.equals("."))
|
||||
.toList();
|
||||
return hosts;
|
||||
}
|
||||
case Lookup.TYPE_NOT_FOUND -> {
|
||||
Lookup a = new Lookup(domain, Type.A);
|
||||
a.run();
|
||||
return a.getResult() == Lookup.SUCCESSFUL ? List.of(domain) : List.of();
|
||||
}
|
||||
case Lookup.HOST_NOT_FOUND -> {
|
||||
return List.of();
|
||||
}
|
||||
default -> throw new TemporaryDnsException(lookup.getErrorString());
|
||||
}
|
||||
} catch (TextParseException e) {
|
||||
return List.of();
|
||||
}
|
||||
}
|
||||
|
||||
/** A message written to the wire byte for byte as received (keeps the DKIM signature valid). */
|
||||
static final class RawMessage extends MimeMessage {
|
||||
private final byte[] raw;
|
||||
|
||||
RawMessage(Session session, byte[] raw) throws MessagingException {
|
||||
super(session, new ByteArrayInputStream(raw));
|
||||
this.raw = raw;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void writeTo(OutputStream os, String[] ignoreHeaders) throws IOException {
|
||||
os.write(raw);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void writeTo(OutputStream os) throws IOException {
|
||||
os.write(raw);
|
||||
}
|
||||
|
||||
@Override
|
||||
public int getSize() {
|
||||
return raw.length;
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user