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 public class SMTP extends MessageProtocol {
  13
  14     // FIXME
  15     private static final Mailbox outgoing = null;
  16
  17     static { new Thread() { public void run() { Outgoing.runq(); } }.start(); }
  18     public static String convdir = null;
  19     public static void main(String[] s) throws Exception {
  20         String logto = System.getProperty("ibex.mail.root", File.separatorChar + "var" + File.separatorChar + "org.ibex.mail");
  21         logto += File.separatorChar + "log";
  22         Log.file(logto);
  23         convdir = System.getProperty("ibex.mail.root", File.separatorChar + "var" + File.separatorChar + "org.ibex.mail");
  24         convdir += File.separatorChar + "conversation";
  25         new File(convdir).mkdirs();
  26         final SMTP smtp = new SMTP();
  27         int port = Integer.parseInt(System.getProperty("ibex.mail.port", "25"));
  28         Log.info(SMTP.class, "binding to port " + port + "...");
  29         ServerSocket ss = new ServerSocket(port);
  30         Log.info(SMTP.class, "listening for connections...");
  31         while(true) {
  32             final Socket sock = ss.accept();
  33             new Thread() { public void run() { smtp.handle(sock); } }.start();
  34         }
  35     }
  36
  37     //public SMTP() { setProtocolName("SMTP"); }
  38     //public ServerRequest createRequest(Connection conn) { return new Listener((TcpConnection)conn); }
  39
  40     public void handle(Socket s) { new Listener(s, "megacz.com").handleRequest(); }
  41
  42     //  FEATURE: exponential backoff on retry time?
  43     public static class Outgoing {
  44         private static final HashSet deadHosts = new HashSet();
  45         private static final org.ibex.util.Queue queue = new org.ibex.util.Queue(100);
  46
  47         public static void send(Message m) throws IOException {
  48             if (m.traces.length >= 100) {
  49                 Log.warn(SMTP.Outgoing.class,
  50                          "Message with " + m.traces.length + " trace hops; silently dropping\n" + m.summary());
  51                 return;
  52             }
  53             synchronized(Outgoing.class) {
  54                 outgoing.add(m);
  55                 queue.append(m);
  56                 Outgoing.class.notify();
  57             }
  58         }
  59
  60         // FIXME!!! ignores more than one destination envelope!!!!
  61         private static boolean attempt(Message m) throws IOException {
  62             InetAddress[] mx = getMailExchangerIPs(m.envelopeTo[0].host);
  63             if (mx.length == 0) {
  64                 Log.warn(SMTP.Outgoing.class, "could not resolve " + m.envelopeTo[0].host + "; bouncing it\n" + m.summary());
  65                 send(m.bounce("could not resolve " + m.envelopeTo[0].host));
  66                 return true;
  67             }
  68             if (new Date().getTime() - m.arrival.getTime() > 1000 * 60 * 60 * 24 * 5) {
  69                 Log.warn(SMTP.Outgoing.class, "could not send message after 5 days; bouncing it\n" + m.summary());
  70                 send(m.bounce("could not send for 5 days"));
  71                 return true;
  72             }
  73             for(int i=0; i<mx.length; i++) {
  74                 if (deadHosts.contains(mx[i])) continue;
  75                 if (attempt(m, mx[i])) { return true; }
  76             }
  77             return false;
  78         }
  79
  80         private static void check(String s) throws IOException {
  81             if (s.startsWith("4") || s.startsWith("5")) throw new IOException("SMTP Error: " + s); }
  82         private static boolean attempt(Message m, InetAddress mx) {
  83             try {
  84                 String vhost = InetAddress.getLocalHost().getHostName();
  85                 String cid = getConversation();
  86                 PrintWriter logf = new PrintWriter(new OutputStreamWriter(new FileOutputStream(convdir+File.separatorChar+cid)));
  87                 Log.setThreadAnnotation("[outgoing smtp: " + mx + " / " + cid + "] ");
  88                 Log.info(SMTP.Outgoing.class, "connecting...");
  89                 Socket s = new Socket(mx, 25);
  90                 Log.info(SMTP.Outgoing.class, "connected");
  91                 LineReader  r = new LoggedLineReader(new InputStreamReader(s.getInputStream()), logf);
  92                 PrintWriter w = new LoggedPrintWriter(new OutputStreamWriter(s.getOutputStream()), logf);
  93                                                                   check(r.readLine());  // banner
  94                 w.print("HELO " + vhost + "\r\n");                check(r.readLine());
  95                 w.print("MAIL FROM: " + m.envelopeFrom + "\r\n"); check(r.readLine());
  96                 w.print("RCPT TO: " + m.envelopeTo + "\r\n");     check(r.readLine());
  97                 w.print("DATA\r\n");                              check(r.readLine());
  98                 w.print(m.body);
  99                 w.print(".\r\n");
 100                 check(r.readLine());
 101                 Log.info(SMTP.Outgoing.class, "message accepted by " + mx);
 102                 // FIXME!
 103                 //outgoing.delete(m);
 104                 s.close();
 105                 return true;
 106             } catch (Exception e) {
 107                 Log.warn(SMTP.Outgoing.class, "unable to send; error=" + e);
 108                 Log.warn(SMTP.Outgoing.class, e);
 109                 return false;
 110             } finally {
 111                 Log.setThreadAnnotation("[outgoing smtp] ");
 112             }
 113         }
 114
 115         static void runq() {
 116             try {
 117                 Log.setThreadAnnotation("[outgoing smtp] ");
 118                 Log.info(SMTP.Outgoing.class, "outgoing thread started; " + outgoing.count(Query.all()) + " messages to send");
 119                 for(Mailbox.Iterator it = outgoing.iterator(); it.cur() != null; it.next()) queue.append(it.cur());
 120                 while(true) {
 121                     int num = queue.size();
 122                     for(int i=0; i<num; i++) {
 123                         Message next = (Message)queue.remove(true);
 124                         boolean good = false;
 125                         try {
 126                             good = attempt(next);
 127                         } catch (IOException e) {
 128                             Log.error(SMTP.Outgoing.class, e);
 129                         } finally {
 130                             if (!good) queue.append(next);
 131                         }
 132                     }
 133                     synchronized(Outgoing.class) {
 134                         Log.info(SMTP.Outgoing.class, "outgoing thread going to sleep");
 135                         Outgoing.class.wait(10 * 60 * 1000);
 136                         deadHosts.clear();
 137                         Log.info(SMTP.Outgoing.class, "outgoing thread woke up; " + queue.size() + " messages in queue");
 138                     }
 139                 }
 140             } catch (Exception e) {
 141                 Log.error(SMTP.Outgoing.class, e);
 142             }
 143         }
 144     }
 145
 146     private static String lastTime = null;
 147     private static int lastCounter = 0;
 148
 149     private class Listener extends Incoming /*implements ServerRequest*/ {
 150         Socket conn;
 151         String vhost;
 152         public void init() { }
 153         public Listener(Socket conn, String vhost) { this.vhost = vhost; this.conn = conn; }
 154
 155         //TcpConnection conn;
 156         //public Listener(TcpConnection conn) { this.conn = conn; conn.getSocket().setSoTimeout(5 * 60 * 1000); }
 157         public boolean handleRequest() {
 158             try {
 159                 conn.setSoTimeout(5 * 60 * 1000);
 160                 StringBuffer logMessage = new StringBuffer();
 161                 String cid = getConversation();
 162                 Log.setThreadAnnotation("[conversation " + cid + "] ");
 163                 InetSocketAddress remote = (InetSocketAddress)conn.getRemoteSocketAddress();
 164                 Log.info(this, "connection from "+remote.getHostName()+":"+remote.getPort()+" ("+remote.getAddress()+")");
 165                 PrintWriter logf = new PrintWriter(new OutputStreamWriter(new FileOutputStream(convdir+File.separatorChar+cid)));
 166                 try {
 167                     return handleRequest(new LoggedLineReader(new InputStreamReader(conn.getInputStream()), logf),
 168                                          new LoggedPrintWriter(new OutputStreamWriter(conn.getOutputStream()), logf));
 169                 } catch(Throwable t) { Log.warn(this, t);
 170                 } finally {            logf.close(); Log.setThreadAnnotation("");
 171                 }
 172             } catch (Exception e) { Log.error(this, e); }
 173             return false;
 174         }
 175
 176         public boolean handleRequest(LineReader rs, PrintWriter ws) throws IOException, MailException {
 177             //ReadStream rs = conn.getReadStream();
 178             //WriteStream ws = conn.getWriteStream();
 179             //ws.setNewLineString("\r\n");
 180             ws.println("220 " + vhost + " ESMTP " + this.getClass().getName());
 181             Address from = null;
 182             Vector to = new Vector();
 183             while(true) {
 184                 String command = rs.readLine();
 185                 String c = command.toUpperCase();
 186                 if (c.startsWith("HELO")) {
 187                     ws.println("250 HELO " + vhost);
 188                     from = null;
 189                     to = new Vector();
 190
 191                 } else if (c.startsWith("EHLO")) {
 192                     ws.println("250-" + vhost);
 193                     ws.println("250-SIZE");
 194                     ws.println("250 PIPELINING");
 195                     from = null;
 196                     to = new Vector();
 197
 198                 } else if (c.startsWith("RSET")) {
 199                     from = null;
 200                     to = new Vector();
 201                     ws.println("250 reset ok");
 202
 203                 } else if (c.startsWith("MAIL FROM:")) {
 204                     command = command.substring(10).trim();
 205                     from = new Address(command);
 206                     ws.println("250 " + from + " is syntactically correct");
 207
 208                 } else if (c.startsWith("RCPT TO:")) {
 209                     if (from == null) {
 210                         ws.println("503 MAIL FROM must precede RCPT TO");
 211                         continue;
 212                     }
 213                     command = command.substring(9).trim();
 214                     if(command.indexOf(' ') != -1) command = command.substring(0, command.indexOf(' '));
 215                     Address addr = new Address(command);
 216                     InetAddress[] mx = getMailExchangerIPs(addr.host);
 217                     to.addElement(addr);
 218                     if (((InetSocketAddress)conn.getRemoteSocketAddress()).getAddress().isLoopbackAddress()) {
 219                         ws.println("250 you are connected locally, so I will let you send");
 220                     } else {
 221                         boolean good = false;
 222                         for(int i=0; !good && i<mx.length; i++)
 223                             if (NetworkInterface.getByInetAddress(mx[i]) != null)
 224                                 good = true;
 225                         if (!good) {
 226                             ws.println("551 sorry, " + addr + " is not on this machine");
 227                             return false;
 228                         }
 229                         ws.println("250 " + addr + " is on this machine; I will deliver it");
 230                     }
 231
 232                 } else if (c.startsWith("DATA")) {
 233                     if (from == null) { ws.println("503 MAIL FROM command must precede DATA"); continue; }
 234                     if (to == null) { ws.println("503 RCPT TO command must precede DATA"); continue; }
 235                     ws.println("354 Enter message, ending with \".\" on a line by itself");
 236                     try {
 237                         Address[] toArr = new Address[to.size()];
 238                         to.copyInto(toArr);
 239                         accept(new Message(from, toArr, new DotTerminatedLineReader(rs)));
 240                         ws.println("250 message accepted");
 241                     } catch (MailException.Malformed mfe) {   ws.println("501 " + mfe.toString());
 242                     } catch (MailException.MailboxFull mbf) { ws.println("452 " + mbf);
 243                     } catch (IOException ioe) {               ws.println("554 " + ioe.toString());
 244                     }
 245                     break;
 246
 247                 } else if (c.startsWith("HELP")) { ws.println("214 you are beyond help.  see a trained professional.");
 248                 } else if (c.startsWith("VRFY")) { ws.println("252 We don't VRFY; proceed anyway");
 249                 } else if (c.startsWith("EXPN")) { ws.println("550 EXPN not available");
 250                 } else if (c.startsWith("NOOP")) { ws.println("250 OK");
 251                 } else if (c.startsWith("QUIT")) { ws.println("221 " + vhost + " closing connection"); break;
 252                 } else                           { ws.println("500 unrecognized command");
 253                 }
 254
 255             }
 256             return false; // always tell resin to close the connection
 257         }
 258     }
 259
 260     private static class DotTerminatedLineReader extends LineReader {
 261         private final LineReader r;
 262         private boolean done = false;
 263         public DotTerminatedLineReader(LineReader r) { super(null); this.r = r; }
 264         public String readLine() throws IOException {
 265             if (done) return null;
 266             String s = r.readLine();
 267             if (s.equals(".")) { done = true; return null; }
 268             if (s.startsWith(".")) return s.substring(1);
 269             return s;
 270         }
 271     }
 272
 273     public static InetAddress[] getMailExchangerIPs(String hostName) {
 274         InetAddress[] ret;
 275         try {
 276             Hashtable env = new Hashtable();
 277             env.put("java.naming.factory.initial", "com.sun.jndi.dns.DnsContextFactory");
 278             DirContext ictx = new InitialDirContext(env);
 279             Attributes attrs = ictx.getAttributes(hostName, new String[] { "MX" });
 280             Attribute attr = attrs.get("MX");
 281             if (attr == null) {
 282                 ret = new InetAddress[1];
 283                 try {
 284                     ret[0] = InetAddress.getByName(hostName);
 285                     return ret;
 286                 } catch (UnknownHostException uhe) {
 287                     Log.warn(SMTP.class, "no MX hosts or A record for " + hostName);
 288                     return new InetAddress[0];
 289                 }
 290             } else {
 291                 ret = new InetAddress[attr.size()];
 292                 NamingEnumeration ne = attr.getAll();
 293                 for(int i=0; ne.hasMore(); i++) {
 294                     String mx = (String)ne.next();
 295                     // FIXME we should be sorting here
 296                     mx = mx.substring(mx.indexOf(" ") + 1);
 297                     if (mx.charAt(mx.length() - 1) == '.') mx = mx.substring(0, mx.length() - 1);
 298                     ret[i] = InetAddress.getByName(mx);
 299                 }
 300             }
 301         } catch (Exception e) {
 302             Log.warn(SMTP.class, "couldn't find MX host for " + hostName + " due to");
 303             Log.warn(SMTP.class, e);
 304             return new InetAddress[0];
 305         }
 306         return ret;
 307     }
 308
 309     private static class LoggedLineReader extends LineReader {
 310         PrintWriter log;
 311         public LoggedLineReader(Reader r, PrintWriter log) { super(r); this.log = log; }
 312         public String readLine() throws IOException {
 313             String s = super.readLine();
 314             if (s != null) { log.println("C: " + s); log.flush(); }
 315             return s;
 316         }
 317     }
 318
 319     private static class LoggedPrintWriter extends PrintWriter {
 320         PrintWriter log;
 321         public LoggedPrintWriter(Writer w, PrintWriter log) { super(w); this.log = log; }
 322         public void println(String s) {
 323             log.println("S: " + s);
 324             super.println(s);
 325             flush();
 326         }
 327     }
 328
 329     static String getConversation() {
 330         String time = new SimpleDateFormat("yy.MMM.dd-hh:mm:ss").format(new Date());
 331         synchronized (SMTP.class) {
 332             if (lastTime != null && lastTime.equals(time)) {
 333                 time += "." + (++lastCounter);
 334             } else {
 335                 lastTime = time;
 336             }
 337         }
 338         return time;
 339     }
 340
 341 }