(* (c) 2017 Hannes Mehnert, all rights reserved *) open Lwt.Infix open Astring open Vmm_core let my_version = `WV0 let command = ref 1 let t : (Lwt_unix.file_descr * Lwt_unix.sockaddr * string) IM.t ref = ref IM.empty module S = struct type t = Lwt_unix.sockaddr let compare : Lwt_unix.sockaddr -> Lwt_unix.sockaddr -> int = compare end module SM = Map.Make(S) let count : int SM.t ref = ref SM.empty let dec s = match SM.find s !count with | exception Not_found -> `Not_found | 1 -> count := SM.remove s !count ; `Close | x -> count := SM.add s (pred x) !count ; `Continue let known_vms : string list ref = ref [] module P = struct let p vm ?(typ = `Counter) name help value = let t_s = function `Counter -> "counter" | `Gauge -> "gauge" in let name = vm ^ "_" ^ name in let p a v = String.concat ~sep:" " [ "#" ; a ; name ; v ] in String.concat ~sep:"\n" [ p "HELP" help ; p "TYPE" (t_s typ) ; name ^ " " ^ value ] let tv (sec, usec) = Printf.sprintf "%Lu.%06d" sec usec let i64 = Int64.to_string let encode_ru vm ru = let p = p vm in String.concat ~sep:"\n" [ p "utime" "user time used" (tv ru.utime) ; p "stime" "system time used" (tv ru.stime) ; p "maxrss" "maximum resident set" (i64 ru.maxrss) ; p ~typ:`Gauge "ixrss" "shared memory" (i64 ru.ixrss) ; p ~typ:`Gauge "idrss" "unshared data" (i64 ru.idrss) ; p ~typ:`Gauge "isrss" "unshared stack" (i64 ru.isrss) ; p "minflt" "page reclaims" (i64 ru.minflt) ; p "maxflt" "page faults" (i64 ru.majflt) ; p "nswap" "swaps" (i64 ru.nswap) ; p "inblock" "block input ops" (i64 ru.inblock) ; p "outblock" "block output ops" (i64 ru.outblock) ; p "msgsnd" "messages send" (i64 ru.msgsnd) ; p "msgrcv" "messages received" (i64 ru.msgrcv) ; p "nsignals" "signals received" (i64 ru.nsignals) ; p "nvcsw" "voluntary context switches" (i64 ru.nvcsw) ; p "nivcsw" "involuntary context switches" (i64 ru.nivcsw) ] let encode_vmm vm xs = let p = p vm in let massage s = let cutted = match String.cut ~sep:"umber of " s with | Some (_, r) -> r | None -> s in let cutted = match String.cut ~sep:"[" cutted with | None -> cutted | Some (l, r) -> match String.cut ~sep:"]" r with | None -> cutted | Some (l', r) when r = "" -> l ^ "_" ^ l' | Some (l', r') -> l ^ "_" ^ l' ^ "_" ^ r' in let cutted = List.fold_left (fun str sep -> match String.cut ~sep str with | None -> str | Some (l, r) -> l ^ r) cutted [ "%" ; "/" ; "-" ] in String.concat ~sep:"_" (String.cuts ~sep:" " cutted) in String.concat ~sep:"\n" (List.map (fun (k, v) -> p (massage k) k (i64 v)) xs) let i32 = Int32.to_string let encode_if vm ifd = let p = p (vm ^ "_" ^ ifd.name) in String.concat ~sep:"\n" (* TODO: flags *) [ p ~typ:`Gauge "send_length" "length of send queue" (i32 ifd.send_length) ; p "max_send_length" "maximum length of send queue" (i32 ifd.max_send_length) ; p "send_drops" "drops in send queue" (i32 ifd.send_drops) ; p ~typ:`Gauge "mtu" "maximum transmission unit" (i32 ifd.mtu) ; p ~typ:`Gauge "baudrate" "linespeed" (i64 ifd.baudrate) ; p "vm_to_host_packets" "packets from vm" (i64 ifd.input_packets) ; p "vm_to_host_errors" "packet errors from vm" (i64 ifd.input_errors) ; p "vm_to_host_bytes" "bytes from vm" (i64 ifd.input_bytes) ; p "vm_to_host_mcast" "packets from vm via multicast" (i64 ifd.input_mcast) ; p "vm_to_host_dropped" "packets dropped from vm" (i64 ifd.input_dropped) ; p "collisions" "collisions on csma interface" (i64 ifd.collisions) ; p "host_to_vm_packets" "packets to vm" (i64 ifd.output_packets) ; p "host_to_vm_errors" "packet errors to vm" (i64 ifd.output_errors) ; p "host_to_vm_bytes" "bytes to vm" (i64 ifd.output_bytes) ; p "host_to_vm_mcast" "packets to vm via multicast" (i64 ifd.output_mcast) ; p "host_to_vm_dropped" "packets dropped to vm" (i64 ifd.output_dropped) ] end (* just a reminder whether we already sent the initial "info" or not *) let f_done = ref false let process db tls hdr data = let open Vmm_wire in let open Rresult.R.Infix in if not (version_eq hdr.version my_version) then begin Logs.err (fun m -> m "unknown wire protocol version") ; Lwt.return_unit end else match hdr.tag with | x when x = Client.log_msg_tag && not !f_done -> f_done := true ; (* issue initial "info" to get all the vm names *) let out = Vmm_wire.Client.cmd `Info !command my_version in command := succ !command ; Logs.debug (fun m -> m "writing %a over TLS" Cstruct.hexdump_pp (Cstruct.of_string out)) ; (Vmm_tls.write_tls tls out >|= function | Ok () -> () | Error _ -> Logs.err (fun m -> m "error while writing") ; ()) | _ -> let r = match hdr.tag with | x when x = Client.log_msg_tag -> Client.decode_log data >>= fun (hdr, event) -> let nam = translate_serial db hdr.Vmm_core.Log.name in begin match event with | `VM_start _ -> known_vms := nam :: !known_vms | `VM_stop _ -> known_vms := List.filter (fun m -> m <> nam) !known_vms | _ -> () end ; Ok `None | x when x = Client.info_msg_tag -> Client.decode_info data >>= fun vms -> let vms = List.map (fun (name, _, _, _) -> translate_serial db name) vms in known_vms := vms ; Ok `None | x when x = Client.stat_msg_tag -> Client.decode_stat data >>= fun (ru, vmm, ifd) -> begin match IM.find hdr.id !t with | exception Not_found -> Logs.err (fun m -> m "unexpected reply") ; Ok `None | (fd, s, vm) -> t := IM.remove hdr.id !t ; let out = String.concat ~sep:"\n" (P.encode_ru vm ru :: P.encode_vmm vm vmm :: List.map (P.encode_if vm) ifd @ [""]) in Ok (`Stat (fd, s, out)) end | x when x = fail_tag -> let res = match IM.find hdr.id !t with | exception Not_found -> `None | (fd, s, _) -> `Sockaddr (fd, s) in t := IM.remove hdr.id !t ; decode_str data >>= fun (msg, _) -> Logs.err (fun m -> m "failed %s" msg) ; Ok res | x -> Rresult.R.error_msgf "ignoring header tag %02X" x in let d (fd, s) = match dec s with | `Continue -> Lwt.return_unit | `Close -> Lwt_unix.close fd | `Not_found -> Logs.err (fun m -> m "sockaddr not found") ; Lwt.return_unit in let open Lwt.Infix in match r with | Ok `None -> Lwt.return_unit | Ok (`Sockaddr s) -> d s | Ok (`Stat (fd, s, out)) -> (Vmm_lwt.write_raw fd out >>= function | Ok () -> d (fd, s) | Error _ -> Logs.err (fun m -> m "exception while writing") ; Lwt.return_unit) | Error (`Msg msg) -> Logs.err (fun m -> m "error while processing: %s" msg) ; Lwt.return_unit let rec tls_listener db tls = (Vmm_tls.read_tls tls >>= function | Error (`Msg msg) -> Logs.err (fun m -> m "error while reading %s" msg) ; Lwt.return (Ok ()) | Error _ -> Logs.err (fun m -> m "received exception in read_tls") ; Lwt.return (Error ()) | Ok (hdr, data) -> process db tls hdr data >>= fun () -> Lwt.return (Ok ())) >>= function | Ok () -> tls_listener db tls | Error () -> Lwt.return_unit let hdr = String.concat ~sep:"\r\n" [ "HTTP/1.1 200 OK" ; "Content-Type: text/plain; version=0.0.4" ; "\r\n" ] (* wait for TCP connection, once received request stats from vmmd, and loop *) let rec tcp_listener db tcp tls = Lwt_unix.accept tcp >>= fun (cs, sockaddr) -> Vmm_lwt.write_raw cs hdr >>= function | Error _ -> Logs.err (fun m -> m "exception while accepting") ; Lwt.return_unit | Ok () -> let l = List.length !known_vms in let ip, port = match sockaddr with Lwt_unix.ADDR_INET (ip, port) -> ip, port | _ -> invalid_arg "unexpected" in Logs.info (fun m -> m "connection from %s:%d with %d known" (Unix.string_of_inet_addr ip) port l) ; (if l = 0 then Lwt_unix.close cs >|= fun () -> Error () else begin count := SM.add sockaddr (List.length !known_vms) !count ; Lwt_list.fold_left_s (fun r vm -> match r with | Error () -> Lwt.return (Error ()) | Ok () -> let vm_id = translate_name db vm in let out = Vmm_wire.Client.cmd `Statistics ~arg:vm_id !command my_version in t := IM.add !command (cs, sockaddr, vm) !t ; command := succ !command ; Vmm_tls.write_tls tls out >|= function | Ok () -> Ok () | Error _ -> Logs.err (fun m -> m "exception while writing") ; Error ()) (Ok ()) !known_vms end) >>= function | Ok () -> tcp_listener db tcp tls | Error () -> Lwt.return_unit let client cas host port cert priv_key db listen_ip listen_port = Nocrypto_entropy_lwt.initialize () >>= fun () -> let auth = if Sys.is_directory cas then `Ca_dir cas else `Ca_file cas in X509_lwt.authenticator auth >>= fun authenticator -> Lwt.catch (fun () -> (* start TCP listening socket *) let tcp = Lwt_unix.(socket PF_INET SOCK_STREAM 0) in Lwt_unix.(setsockopt tcp SO_REUSEADDR true) ; let addr = Lwt_unix.ADDR_INET (Ipaddr_unix.V4.to_inet_addr listen_ip, listen_port) in Lwt_unix.bind tcp addr >>= fun () -> Lwt_unix.listen tcp 1 ; (* setup remote connection to VMMD *) Lwt_unix.gethostbyname host >>= fun host_entry -> let host_inet_addr = Array.get host_entry.Lwt_unix.h_addr_list 0 in let fd = Lwt_unix.socket host_entry.Lwt_unix.h_addrtype Lwt_unix.SOCK_STREAM 0 in Lwt_unix.connect fd (Lwt_unix.ADDR_INET (host_inet_addr, port)) >>= fun _ -> X509_lwt.private_of_pems ~cert ~priv_key >>= fun cert -> let certificates = `Single cert in let client = Tls.Config.client ~reneg:true ~certificates ~authenticator () in Tls_lwt.Unix.client_of_fd client (* ~host *) fd >>= fun tls -> (* loop on both tcp and tls connections *) Lwt.join [ tcp_listener db tcp tls ; tls_listener db tls ]) (fun exn -> Logs.err (fun m -> m "failed to establish TLS connection: %s" (Printexc.to_string exn)) ; Lwt.return_unit) let run_client _ cas cert key (host, port) db listen_ip listen_port = Printexc.register_printer (function | Tls_lwt.Tls_alert x -> Some ("TLS alert: " ^ Tls.Packet.alert_type_to_string x) | Tls_lwt.Tls_failure f -> Some ("TLS failure: " ^ Tls.Engine.string_of_failure f) | _ -> None) ; Sys.(set_signal sigpipe Signal_ignore) ; let db = let open Rresult.R.Infix in match db with | None -> [] | Some db -> match Bos.OS.File.read_lines (Fpath.v db) >>= parse_db with | Ok db -> db | Error (`Msg m) -> Logs.warn (fun f -> f "couldn't parse database %s" m) ; [] in Lwt_main.run (client cas host port cert key db listen_ip listen_port) let setup_log style_renderer level = Fmt_tty.setup_std_outputs ?style_renderer (); Logs.set_level level; Logs.set_reporter (Logs_fmt.reporter ~dst:Format.std_formatter ()) open Cmdliner let setup_log = Term.(const setup_log $ Fmt_cli.style_renderer () $ Logs_cli.level ()) let host_port : (string * int) Arg.converter = let parse s = match String.cut ~sep:":" s with | None -> `Error "broken: no port specified" | Some (hostname, port) -> try `Ok (hostname, int_of_string port) with Not_found -> `Error "failed to parse port" in parse, fun ppf (h, p) -> Format.fprintf ppf "%s:%d" h p let cas = let doc = "The full path to PEM encoded certificate authorities. Can either be a FILE or a DIRECTORY." in Arg.(required & pos 0 (some string) None & info [] ~docv:"FILE" ~doc) let client_cert = let doc = "Use a client certificate chain" in Arg.(required & pos 1 (some file) None & info [] ~doc) let client_key = let doc = "Use a client key" in Arg.(required & pos 2 (some file) None & info [] ~doc) let destination = Arg.(required & pos 3 (some host_port) None & info [] ~docv:"destination" ~doc:"the destination hostname:port to connect to") let ip : Ipaddr.V4.t Arg.converter = let parse s = try `Ok (Ipaddr.V4.of_string_exn s) with Not_found -> `Error "broken" in parse, Ipaddr.V4.pp_hum let address = let doc = "Address to listen on" in Arg.(value & opt ip (Ipaddr.V4.of_string_exn "127.0.0.1") & info [ "address" ] ~doc) let port = let doc = "TCP port to listen on" in Arg.(value & opt int 9080 & info [ "port" ] ~doc) let db = let doc = "Certificate database" in Arg.(value & opt (some file) None & info [ "db" ] ~doc) let cmd = let doc = "VMM TLS client" in let man = [ `S "DESCRIPTION" ; `P "$(tname) connects to a server and initiates a TLS handshake" ] in Term.(pure run_client $ setup_log $ cas $ client_cert $ client_key $ destination $ db $ address $ port), Term.info "vmm_client" ~version:"%%VERSION_NUM%%" ~doc ~man let () = match Term.eval cmd with `Error _ -> exit 1 | _ -> exit 0