src/org/ibex/mail/protocol/SMTP.java

   1 package org.ibex.mail.protocol;
   2 import org.ibex.mail.*;
   3 import org.ibex.mail.target.*;
   4 import org.ibex.util.*;
   5 import java.net.*;
   6 import java.io.*;
   7 import java.util.*;
   8 import java.text.*;
   9 import javax.naming.*;
  10 import javax.naming.directory.*;
  11
  12 // FEATURE: exponential backoff on retry time?
  13 public class SMTP {
  14
  15     private static final Mailbox spool =
  16         FileBasedMailbox.getFileBasedMailbox(Mailbox.STORAGE_ROOT, false).slash("spool", true).slash("smtp", true);
  17     static { new Thread() { public void run() { Outgoing.runq(); } }.start(); }
  18
  19     // Server //////////////////////////////////////////////////////////////////////////////
  20
  21     public static class Server extends Connection {
  22         public Server(Socket conn, String vhost) throws IOException { super(conn, vhost); }
  23         public boolean handleRequest() throws IOException, MailException {
  24             conn.setSoTimeout(5 * 60 * 1000);
  25             println("220 " + vhost + " ESMTP " + this.getClass().getName());
  26             Address from = null;
  27             Vector to = new Vector();
  28             for(String command = readln(); ; command = readln()) {
  29                 String c = command.toUpperCase();
  30                 if (c.startsWith("HELO"))        { println("250 HELO " + vhost);                  from = null; to = new Vector();
  31                 } else if (c.startsWith("EHLO")) { println("250-\r\n250-SIZE\r\n250 PIPELINING"); from = null; to = new Vector();
  32                 } else if (c.startsWith("RSET")) { println("250 reset ok");                       from = null; to = new Vector();
  33                 } else if (c.startsWith("HELP")) { println("214 you are beyond help.  see a trained professional.");
  34                 } else if (c.startsWith("VRFY")) { println("252 We don't VRFY; proceed anyway");
  35                 } else if (c.startsWith("EXPN")) { println("550 EXPN not available");
  36                 } else if (c.startsWith("NOOP")) { println("250 OK");
  37                 } else if (c.startsWith("QUIT")) { println("221 " + vhost + " closing connection"); return false;
  38                 } else if (c.startsWith("MAIL FROM:")) {
  39                     println("250 " + (from = new Address(command.substring(10).trim())) + " is syntactically correct");
  40                 } else if (c.startsWith("RCPT TO:")) {
  41                     if (from == null) { println("503 MAIL FROM must precede RCPT TO"); continue; }
  42                     command = command.substring(9).trim();
  43                     if(command.indexOf(' ') != -1) command = command.substring(0, command.indexOf(' '));
  44                     Address addr = new Address(command);
  45                     InetAddress[] mx = getMailExchangerIPs(addr.host);
  46                     to.addElement(addr);
  47                     if (((InetSocketAddress)conn.getRemoteSocketAddress()).getAddress().isLoopbackAddress()) {
  48                         println("250 you are connected locally, so I will let you send");
  49                     } else {
  50                         boolean good = false;
  51                         for(int i=0; !good && i<mx.length; i++) if (NetworkInterface.getByInetAddress(mx[i]) != null) good = true;
  52                         if (!good) { println("551 sorry, " + addr + " is not on this machine"); return false; }
  53                         println("250 " + addr + " is on this machine; I will deliver it");
  54                     }
  55                 } else if (c.startsWith("DATA")) {
  56                     if (from == null) { println("503 MAIL FROM command must precede DATA"); continue; }
  57                     if (to == null) { println("503 RCPT TO command must precede DATA"); continue; }
  58                     println("354 Enter message, ending with \".\" on a line by itself");
  59                     try {
  60                         for(int i=0; i<to.size(); i++)
  61                             Target.root.accept(new Message(from, (Address)to.elementAt(i), new DotTerminatedLineReader(lr)));
  62                         println("250 message accepted");
  63                     } catch (MailException.Malformed mfe) {   println("501 " + mfe.toString());
  64                     } catch (MailException.MailboxFull mbf) { println("452 " + mbf);
  65                     } catch (IOException ioe) {               println("554 " + ioe.toString());
  66                     }
  67                     return false;
  68                 } else                           { println("500 unrecognized command"); }
  69             }
  70         }
  71     }
  72
  73
  74     // Outgoing Mail Thread //////////////////////////////////////////////////////////////////////////////
  75
  76     public static class Outgoing {
  77         private static final HashSet deadHosts = new HashSet();
  78         public static void send(Message m) throws IOException {
  79             if (m.traces.length >= 100)
  80                 Log.warn(SMTP.Outgoing.class, "Message with " + m.traces.length + " trace hops; dropping\n" + m.summary());
  81             else synchronized(Outgoing.class) {
  82                 spool.add(m);
  83                 Outgoing.class.notify();
  84             }
  85         }
  86
  87         private static boolean attempt(Message m) throws IOException {
  88             InetAddress[] mx = getMailExchangerIPs(m.envelopeTo.host);
  89             if (mx.length == 0) {
  90                 Log.warn(SMTP.Outgoing.class, "could not resolve " + m.envelopeTo.host + "; bouncing it\n" + m.summary());
  91                 send(m.bounce("could not resolve " + m.envelopeTo.host));
  92                 return true;
  93             }
  94             if (new Date().getTime() - m.arrival.getTime() > 1000 * 60 * 60 * 24 * 5) {
  95                 Log.warn(SMTP.Outgoing.class, "could not send message after 5 days; bouncing it\n" + m.summary());
  96                 send(m.bounce("could not send for 5 days"));
  97                 return true;
  98             }
  99             for(int i=0; i<mx.length; i++) {
 100                 if (deadHosts.contains(mx[i])) continue;
 101                 if (attempt(m, mx[i])) { return true; }
 102             }
 103             return false;
 104         }
 105
 106         private static void check(String s) { if (s.startsWith("4")||s.startsWith("5")) throw new MailException(s); }
 107         private static boolean attempt(final Message m, final InetAddress mx) {
 108             boolean accepted0 = false;
 109             try {
 110                 Log.info(SMTP.Outgoing.class, "connecting...");
 111                 accepted0 = new Connection(new Socket(mx, 25), InetAddress.getLocalHost().getHostName()) {
 112                     public boolean handleRequest() throws IOException {
 113                         Log.info(SMTP.Outgoing.class, "connected");
 114                         check(readln());  // banner
 115                         println("HELO " + vhost );                check(readln());
 116                         println("MAIL FROM: " + m.envelopeFrom);  check(readln());
 117                         println("RCPT TO: " + m.envelopeTo);      check(readln());
 118                         println("DATA");                          check(readln());
 119                         println(m.body);
 120                         println(".");
 121                         check(readln());
 122                         Log.info(SMTP.Outgoing.class, "message accepted by " + mx);
 123                         boolean accepted = true;
 124                         try {
 125                             conn.close();
 126                         } finally {
 127                             return accepted;
 128                         }
 129                     }
 130                 }.handle();
 131                 return accepted0;
 132             } catch (Exception e) {
 133                 if (accepted0) return true;
 134                 Log.warn(SMTP.Outgoing.class, "unable to send; error=" + e);
 135                 Log.warn(SMTP.Outgoing.class, e);
 136                 return false;
 137             } finally { Log.setThreadAnnotation("[outgoing smtp] "); }
 138         }
 139
 140         static void runq() {
 141             try {
 142                 Log.setThreadAnnotation("[outgoing smtp] ");
 143                 Log.info(SMTP.Outgoing.class, "outgoing thread started; " + spool.count(Query.all()) + " messages to send");
 144                 while(true) {
 145                     for(Mailbox.Iterator it = spool.iterator(); it.cur() != null; it.next())
 146                         try                   { if (attempt(it.cur())) it.delete(); }
 147                         catch (IOException e) { Log.error(SMTP.Outgoing.class, e); }
 148                     synchronized(Outgoing.class) {
 149                         Log.info(SMTP.Outgoing.class, "outgoing thread going to sleep");
 150                         Outgoing.class.wait(10 * 60 * 1000);
 151                         deadHosts.clear();
 152                         Log.info(SMTP.Outgoing.class,"outgoing thread woke up; "+spool.count(Query.all())+" messages in queue");
 153                     }
 154                 }
 155             } catch (Exception e) { Log.error(SMTP.Outgoing.class, e); }
 156         }
 157     }
 158
 159     public static InetAddress[] getMailExchangerIPs(String hostName) {
 160         InetAddress[] ret;
 161         try {
 162             Hashtable env = new Hashtable();
 163             env.put("java.naming.factory.initial", "com.sun.jndi.dns.DnsContextFactory");
 164             DirContext ictx = new InitialDirContext(env);
 165             Attributes attrs = ictx.getAttributes(hostName, new String[] { "MX" });
 166             Attribute attr = attrs.get("MX");
 167             if (attr == null) {
 168                 ret = new InetAddress[1];
 169                 try {
 170                     ret[0] = InetAddress.getByName(hostName);
 171                     return ret;
 172                 } catch (UnknownHostException uhe) {
 173                     Log.warn(SMTP.class, "no MX hosts or A record for " + hostName);
 174                     return new InetAddress[0];
 175                 }
 176             } else {
 177                 ret = new InetAddress[attr.size()];
 178                 NamingEnumeration ne = attr.getAll();
 179                 for(int i=0; ne.hasMore(); i++) {
 180                     String mx = (String)ne.next();
 181                     // FIXME we should be sorting here
 182                     mx = mx.substring(mx.indexOf(" ") + 1);
 183                     if (mx.charAt(mx.length() - 1) == '.') mx = mx.substring(0, mx.length() - 1);
 184                     ret[i] = InetAddress.getByName(mx);
 185                 }
 186             }
 187         } catch (Exception e) {
 188             Log.warn(SMTP.class, "couldn't find MX host for " + hostName + " due to");
 189             Log.warn(SMTP.class, e);
 190             return new InetAddress[0];
 191         }
 192         return ret;
 193     }
 194
 195
 196     // Helpers //////////////////////////////////////////////////////////////////////////////
 197
 198     public static class DotTerminatedLineReader extends LineReader {
 199         private final LineReader r;
 200         private boolean done = false;
 201         public DotTerminatedLineReader(LineReader r) { super(null); this.r = r; }
 202         public String readLine() throws IOException {
 203             if (done) return null;
 204             String s = r.readLine();
 205             if (s.equals(".")) { done = true; return null; }
 206             if (s.startsWith(".")) return s.substring(1);
 207             return s;
 208         }
 209     }
 210
 211 }