.

lib/hm.ml

@@ -160,5 +160,6 @@ struct
       [
         smtp_verifier_service ~pool:pool0 ?stop ~port stack resolver conf_server
       ; smtp_logic ~pool:pool1 ~info ~tls stack resolver messaged locals dns
+      ; smtp_send_emails ...
       ]
 end

lib/mxs.ml

@@ -1,23 +1,22 @@
-type elt = {
-    preference: int
-  ; mx_ipaddr: Ipaddr.t
-  ; mx_domain: [ `host ] Domain_name.t option
-}
+type key = Dns.Mx.t
 
-let pp_elt : elt Fmt.t =
+let pp_key : key Fmt.t =
  fun ppf elt ->
-  Fmt.pf ppf "{ @[<hov>preference= %d;@ mx_ipaddr= %a;@ mx_domain= %a;@] }"
-    elt.preference Ipaddr.pp elt.mx_ipaddr
-    (Fmt.option Domain_name.pp)
-    elt.mx_domain
+  Fmt.pf ppf "{ @[<hov>preference= %d;@ mail_exchange= %a;@] }"
+    elt.Dns.Mx.preference Domain_name.pp
+    elt.Dns.Mx.mail_exchange
 
-module Elt = struct
-  type t = elt
+module Key = struct
+  type t = key
 
-  let compare {mx_ipaddr= a; _} {mx_ipaddr= b; _} = Ipaddr.compare a b
+  let compare {Dns.Mx.preference= a; _} {Dns.Mx.preference= b; _} = Int.compare a b
 end
 
-include (Set.Make (Elt) : Set.S with type elt := elt)
+include (Map.Make (Key) : Map.S with type key := key)
 
-let v ~preference ?domain ipaddr =
-  {preference; mx_domain= domain; mx_ipaddr= ipaddr}
+let v ~preference ~domain:mail_exchange ipaddr =
+  singleton { preference; mail_exchange } ipaddr
+
+let vs =
+  (Fun.flip List.fold_left empty) begin fun acc (mx, ipaddr) ->
+    add mx ipaddr acc end

lib/mxs.mli

@@ -1,18 +1,8 @@
-(** A domain can have several Mail eXchange services ordered by a preference
-    number. The lower preference is a higher priority. This module provides a
-    way to store them. The uniq identifier for each MX value is the
-    [Ipaddr.V4.t]. *)
+type key = Dns.Mx.t
 
-type elt = {
-    preference: int
-  ; mx_ipaddr: Ipaddr.t
-  ; mx_domain: [ `host ] Domain_name.t option
-}
-(** Type of a MX value. *)
+val pp_key : key Fmt.t
 
-val pp_elt : elt Fmt.t
+include Map.S with type key := key
 
-val v : preference:int -> ?domain:[ `host ] Domain_name.t -> Ipaddr.t -> elt
-(** [v ~preference ?domain mx_ipaddr] returns an MX value. *)
-
-include Set.S with type elt := elt
+val v : preference:int -> domain:[ `host ] Domain_name.t -> Ipaddr.t -> Ipaddr.t t
+val vs : (Dns.Mx.t * Ipaddr.t) list -> Ipaddr.t t

lib/ptt_tuyau.ml

@@ -28,59 +28,33 @@ module Client (Stack : Tcpip.Stack.V4V6) = struct
       ~info
       ~tls
       stack
-      mx_ipaddr
-      emitter
-      producer
-      recipients =
-    Stack.TCP.create_connection stack (mx_ipaddr, 25)
+      ipaddr
+      sender
+      recipients
+      stream =
+    Stack.TCP.create_connection stack (ipaddr, 25)
     >|= R.reword_error (fun err -> `Flow err)
     >>? fun flow ->
     let flow' = Flow.make flow in
     let ctx =
       Sendmail_with_starttls.Context_with_tls.make ?encoder ?decoder ?queue ()
     in
     let domain =
-      let vs = Domain_name.to_strings info.Ptt.Logic.domain in
+      let vs = Domain_name.to_strings info.Ptt.Logic.domain in (* the domain of our SMTP stack => ptt *)
       Colombe.Domain.Domain vs in
     Lwt.catch
       (fun () ->
-        Sendmail_with_starttls.sendmail lwt rdwr flow' ctx tls ~domain emitter
-          recipients producer
+        Sendmail_with_starttls.sendmail lwt rdwr flow' ctx tls ~domain sender
+          recipients stream
         |> Lwt_scheduler.prj
-        >|= R.reword_error (fun err -> `Sendmail err))
       (function
-        | Failure err ->
-          Lwt.return (R.error_msg err)
-          (* XXX(dinosaure): should come from [rdwr]. *)
+        | Failure err -> Lwt.return (R.error_msg err)
         | exn -> Lwt.return (Error (`Exn exn)))
     >>= fun res ->
-    Stack.TCP.close flow >>= fun () ->
-    match res with
-    | Ok () ->
-      Log.debug (fun m ->
-          m "Email to %a was sent!"
-            Fmt.(Dump.list Colombe.Forward_path.pp)
-            recipients);
-      Lwt.return (Ok ())
-    | Error (`Sendmail `STARTTLS_unavailable) ->
-      Lwt.return_error `STARTTLS_unavailable
-    | Error (`Sendmail err) ->
-      Log.err (fun m ->
-          m "Got a sendmail error when we tried to sent to %a: %a"
-            Fmt.(Dump.list Colombe.Forward_path.pp)
-            recipients Sendmail_with_starttls.pp_error err);
-      Lwt.return (R.error_msgf "%a" Sendmail_with_starttls.pp_error err)
-    | Error (`Msg msg) as err ->
-      Log.err (fun m ->
-          m "Got an error when we tried to sent to %a: %s"
-            Fmt.(Dump.list Colombe.Forward_path.pp)
-            recipients msg);
-      Lwt.return err
-    | Error (`Exn exn) ->
-      Lwt.return (R.error_msgf "Unknown error: %s" (Printexc.to_string exn))
+    Stack.TCP.close flow >|= fun () -> res
 
   let sendmail_without_tls
-      ?encoder ?decoder ~info stack mx_ipaddr emitter producer recipients =
+      ?encoder ?decoder ~info stack ipaddr sender recipients stream =
     Stack.TCP.create_connection stack (mx_ipaddr, 25)
     >|= R.reword_error (fun err -> `Flow err)
     >>? fun flow ->
@@ -93,24 +67,11 @@ module Client (Stack : Tcpip.Stack.V4V6) = struct
       (fun () ->
         Sendmail.sendmail lwt rdwr flow' ctx ~domain emitter recipients producer
         |> Lwt_scheduler.prj
-        >|= R.reword_error (fun err -> `Sendmail err))
       (function
         | Failure err -> Lwt.return (R.error_msg err)
         | exn -> Lwt.return (Error (`Exn exn)))
     >>= fun res ->
-    Stack.TCP.close flow >>= fun () ->
-    match res with
-    | Ok () -> Lwt.return (Ok ())
-    | Error (`Sendmail err) ->
-      Lwt.return (R.error_msgf "%a" Sendmail.pp_error err)
-    | Error (`Msg _) as err -> Lwt.return err
-    | Error (`Exn exn) ->
-      Lwt.return (R.error_msgf "Unknown error: %s" (Printexc.to_string exn))
-
-  let pp_error ppf = function
-    | `Msg err -> Fmt.string ppf err
-    | `Flow err -> Stack.TCP.pp_error ppf err
-    | `STARTTLS_unavailable -> Fmt.string ppf "STARTTLS unavailable"
+    Stack.TCP.close flow >|= fun () -> res
 end
 
 module Server (Time : Mirage_time.S) (Stack : Tcpip.Stack.V4V6) = struct

lib/sendmail.ml

@@ -0,0 +1,115 @@
+type recipients =
+  { domain : [ `Ipaddr of Ipaddr.t | `Domain of [ `host ] Domain_name.t ]
+  ; locals : [ `All | `Some of Emile.local list ] }
+
+(* 1: recipients
+ *    | domain: robur.coop
+ *    | locals: `Some [ reynir; dinosaure ] *)
+
+let warn_about_an_unreachable_mail_exchange ~domain ~mail_exchange msg =
+  Log.warn @@ fun m -> m "Impossible to resolve %a, a mail exchange server for %a: %s"
+    Domain_name.pp mail_exchange Domain_name.pp domain msg
+
+let warn_that_starttls_is_unavailable ~domain ~ipaddr =
+  Log.warn @@ fun m -> m "STARTTLS is unavailable for %a (%a)" Domain_name.pp domain Ipaddr.pp ipaddr
+
+type error =
+  [ `No_mail_exchange_servers_for of [ `host ] Domain_name.t ]
+
+(* recipients -> Colombe.Forward_path.t list *)
+let recipients_to_forward_paths ~domain recipients =
+  let open Colombe in
+  let open Forward_path in
+  let local_to_forward_path local =
+    let local = List.map (function `Atom x -> x | `String x -> x) local in
+    { Path.local= `Dot_string local; domain; rest= [] } in
+  match recipients.locals with
+  | `All -> [ Domain domain ]
+  | `Some locals -> Lust.map local_to_forward_path locals
+
+let single_sendmail t stack ipaddr sender recipients stream =
+  sendmail stack ipaddr sender recipients stream >>= function
+  | Error `STARTTLS_unavailable ->
+    warn_that_starttls_is_unavailable ipaddr;
+    sendmail_without_tls stack ipaddr sender recipients stream
+  | Ok () -> Lwt.return_ok ()
+  | Error err -> Lwt.return_error err
+
+(* example:
+ * To: [email protected], [email protected], [email protected]
+ *
+ * Hello World!
+ *
+ * -> incoming email with one stream with "Hello World!"
+ * -> signing
+ * -> our ((stream of DKIM-fieldi) ^ (incoming stream "Hello World")) + recipients
+ *    | [email protected], [email protected], [email protected]
+ * -> aggregate
+ *    | { domain= robur.coop; locals= [ reynir; dinosaure ] }
+ *    | { domain= gmail.com; locals= [ romain.calascibetta ] }
+ * -> multiplex the incoming stream to multiple streams (in our example, 2)
+ * -> Lwt_list.iter push_to_send [ recipients with robur.coop, copied incoming stream
+ *                               ; recipients with gmail.com, copied incoming stream ]
+ *
+ * another thread is: get_emails_to_send
+ * -> Lwt_stream.get incoming_emails_to_send : (recipients * string Lwt_stream.t) Lwt_stream.t
+ * -> Some (recipients, stream)
+ * -> sendmail recipients stream
+ *
+ * MX gmail.com
+ * gmail-smtp-in.l.google.com => A: 108.177.15.27
+ * alt2.gmail-smtp-in.l.google.com => A: 142.251.9.26
+ *)
+
+type t =
+  { stream : _ Lwt_stream.t
+  ; info : Ptt.info }
+
+let sendmail t resolver dns sender (recipients : recipients) (data : string Lwt_stream.t) =
+  let ( let** ) = Lwt_result.bind in
+  let ( let* ) = Lwt.bind in
+  let domain = recipients.domain in
+  let** mx_domain, mxs =
+    match recipients.domain with
+    | `Ipaddr (Ipaddr.V4 v4 as mx_ipaddr) ->
+      Lwt.return_ok (Domain.IPv4 v4, Ptt.Mxs.(v ~preference:0 mx_ipaddr))
+    | `Ipaddr (Ipaddr.V6 v6 as mx_ipaddr) ->
+      Lwt.return_ok (Domain.IPv6 v6, Ptt.Mxs.(v ~preference:0 mx_ipaddr))
+    | `Domain domain ->
+      let* result = resolver.getmxbyname dns host in
+      match result with
+      | Ok mxs ->
+          let mxs = (Fun.flip Lwt_list.fold_left_s (Dns.Mx_set.to_list mxs))
+            begin fun acc ({ Dns.Mx.mail_exchange; _ } as mx) ->
+              resolver.getabyname dns mail_exchange >>= function
+              | Ok ipaddr -> Lwt.return ((mx, ipaddr) :: acc)
+              | Error (`Msg err) ->
+                warn_about_an_unreachable_mail_exchange ~domain ~mail_exchange msg;
+                Lwt.return acc end |> Mxs.vs in
+          Domain.Domain (Domain_name.to_strings domain), mxs in
+      | Error (`Msg err) ->
+        Lwt.return_error (`No_mail_exchange_servers_for domain)
+  let** () =
+    if Mxs.is_empty mxs
+    then Lwt.return_error (`No_mail_exchange_servers_for recipients.domain)
+    else Lwt.return_ok () in
+  let recipients = recipients_to_forward_paths recipients in
+  let mxs = Mxs.to_list mxs in
+  let rec go = function
+    | [] ->
+      let recipients = recipients_of_sender sender in
+      Lwt_stream.push t.stream (recipients, error_sendmail);
+      Lwt.return_unit
+    | ((mx : Dns.Mx.t), ipaddr) :: mxs ->
+      let* result = single_sendmail t stack ~domain ipaddr sender recipients stream in
+      match result with
+      | Ok () -> Lwt.return_ok ()
+      | Error err -> go mxs in
+  go None mxs
+
+let rec smtp_send_emails t resolver dns =
+  Lwt_stream.get t.stream >>= function
+  | Some (sender, recipients, data) ->
+    sendmail t resolver dns sender recipients data >>= fun () ->
+    smtp_send_emails t resolver dns
+  | None -> Lwt.return_unit