Support for SSL in broker

Author: ThomasTJdev

src/nmqtt.nim

@@ -80,7 +80,9 @@ type
   MqttCtx* = ref object
     host: string
     port: Port
-    doSsl: bool
+    sslOn: bool
+    verbosity: int
+    beenConnected: bool
     username: string
     password: string
     state: State
@@ -187,7 +189,9 @@ when defined(broker):
     MqttBroker* = ref object
       host: string
       port: Port
-      doSsl: bool
+      sslOn: bool
+      sslCert: string
+      sslKey: string
       verbosity: int
       connections: Table[string, MqttCtx]
       retained: Table[string, RetainedMsg] # Topic, RetaindMsg
@@ -276,7 +280,7 @@ proc newPkt(typ: PktType=NOTYPE, flags: uint8=0): Pkt =
 #
 proc dmp(ctx: MqttCtx, s: string) =
   when not defined(broker):
-    if defined(dev):
+    if defined(dev) or ctx.verbosity >= 2:
       stderr.write "\e[1;30m" & s & "\e[0m\n"
   when defined(broker):
     if defined(dev) or mqttbroker.verbosity >= 2:
@@ -393,7 +397,8 @@ proc sendDisconnect(ctx: MqttCtx): Future[bool] {.async.}
 proc close(ctx: MqttCtx, reason: string) {.async.} =
   if ctx.state in {Connecting, Connected}:
     ctx.state = Disconnecting
-    ctx.dbg "Closing: " & reason
+    if ctx.verbosity >= 1:
+      ctx.dbg "Closing: " & reason
     discard await ctx.sendDisconnect()
     ctx.s.close()
     ctx.state = Disconnected
@@ -433,7 +438,15 @@ proc recv(ctx: MqttCtx): Future[Pkt] {.async.} =
 
   var r: int
   var b: uint8
-  r = await ctx.s.recvInto(b.addr, b.sizeof)
+
+  # TODO:
+  # When the broker is running in SSL-mode, we need the try/except, since
+  # we will encounter a silent crash in recvInto, when the client is
+  # not actually using SSL.
+  try:
+    r = await ctx.s.recvInto(b.addr, b.sizeof)
+  except:
+    return
   if r != 1:
     when not defined(broker):
       await ctx.close("remote closed connection")
@@ -800,6 +813,7 @@ proc onConnect(ctx: MqttCtx, pkt: Pkt) {.async.} =
 
     mqttbroker.connections[ctx.clientid] = ctx
     ctx.state = Connected
+    ctx.beenConnected = true
     asyncCheck keepAliveMonitor(ctx)
 
     if mqttbroker.verbosity >= 1:
@@ -814,7 +828,9 @@ proc onConnAck(ctx: MqttCtx, pkt: Pkt): Future[void] =
   ctx.state = Connected
   let (code, _) = pkt.getu8(1)
   if code == 0:
-    ctx.dbg "Connection established"
+    ctx.beenConnected = true
+    if ctx.verbosity >= 1:
+      ctx.dbg "Connection established"
   else:
     ctx.wrn "Connect failed, code: " & $code
   result = ctx.work()
@@ -1056,7 +1072,8 @@ proc runRx(ctx: MqttCtx) {.async.} =
         break
       await ctx.handle(pkt)
   except OsError:
-    ctx.wrn "Boom, sending packet to closed socket"
+    if ctx.verbosity >= 2:
+      ctx.wrn "Boom, socket is closed"
 
 proc runPing(ctx: MqttCtx) {.async.} =
   while true:
@@ -1071,10 +1088,11 @@ proc connectBroker(ctx: MqttCtx) {.async.} =
   if ctx.keepAlive == 0:
     ctx.keepAlive = 60
 
-  ctx.dbg "Connecting to " & ctx.host & ":" & $ctx.port
+  if ctx.verbosity >= 1:
+    ctx.dbg "Connecting to " & ctx.host & ":" & $ctx.port
   try:
     ctx.s = await asyncnet.dial(ctx.host, ctx.port)
-    if ctx.doSsl:
+    if ctx.sslOn:
       when defined(ssl):
         ctx.ssl = newContext(protSSLv23, CVerifyNone)
         wrapConnectedSocket(ctx.ssl, ctx.s, handshakeAsClient)
@@ -1087,7 +1105,10 @@ proc connectBroker(ctx: MqttCtx) {.async.} =
       asyncCheck ctx.runRx()
       asyncCheck ctx.runPing()
   except OSError as e:
-    ctx.dbg "Error connecting to " & ctx.host & " " & e.msg
+    if ctx.verbosity >= 1 or not ctx.beenConnected:
+      ctx.dbg "Error connecting to " & ctx.host
+    if ctx.verbosity >= 2:
+      echo e.msg
     ctx.state = Error
 
 proc runConnect(ctx: MqttCtx) {.async.} =
@@ -1125,11 +1146,11 @@ proc set_ping_interval*(ctx: MqttCtx, txInterval: int = 60) =
   if txInterval > 0 and txInterval < 65535:
     ctx.keepAlive = txInterval.uint16
 
-proc set_host*(ctx: MqttCtx, host: string, port: int=1883, doSsl=false) =
+proc set_host*(ctx: MqttCtx, host: string, port: int=1883, sslOn=false) =
   ## Set the MQTT host
   ctx.host = host
   ctx.port = Port(port)
-  ctx.doSsl = doSsl
+  ctx.sslOn = sslOn
 
 proc set_auth*(ctx: MqttCtx, username: string, password: string) =
   ## Set the authentication for the host.
@@ -1144,6 +1165,10 @@ proc set_will*(ctx: MqttCtx, topic, msg: string, qos=0, retain=false) =
   ctx.willQoS    = qos.uint8
   ctx.willRetain = retain
 
+proc set_verbosity*(ctx: MqttCtx, verbosity: int) =
+  ## Set the verbosity.
+  ctx.verbosity = verbosity
+
 proc connect*(ctx: MqttCtx) {.async.} =
   ## Connect to the broker.
   await ctx.connectBroker()
@@ -1158,7 +1183,7 @@ proc start*(ctx: MqttCtx) {.async.} =
 
 proc disconnect*(ctx: MqttCtx) {.async.} =
   ## Disconnect from the broker.
-  await ctx.close("User request")
+  await ctx.close("disconnect")
   ctx.state = Disabled
 
 proc publish*(ctx: MqttCtx, topic: string, message: string, qos=0, retain=false) {.async.} =

src/utils/broker.nim

@@ -18,6 +18,7 @@ proc processClient(s: AsyncSocket) {.async.} =
   let ctx = MqttCtx()
   ctx.s = s
   ctx.state = Connecting
+  ctx.sslOn = mqttbroker.sslOn
 
   while ctx.state in [Connecting, Connected]:
     try:
@@ -34,15 +35,26 @@ proc processClient(s: AsyncSocket) {.async.} =
         ctx.state = Error
       break
 
-  if ctx.state == Error:
+  if not ctx.beenConnected:
+    # When the client hasn't been connected to the broker, then we don't
+    # want to check for LWT, subscriptions etc. - we just want to close
+    # and remove the socket!
+    try:
+      ctx.s.close()
+    except:
+      if mqttbroker.verbosity >= 3 and ctx.sslOn:
+        wrn("Someone is trying to connect, properly without SSL.")
+    return
+
+  if ctx.state == Error and ctx.beenConnected:
     # This happens on a ungraceful disconnects from the client.
     asyncCheck sendWill(ctx)
 
-  if ctx.state in [Disconnected, Error]:
+  if ctx.state in [Disconnected, Error] and ctx.beenConnected:
     # Remove the client from the register for subscribers.
     asyncCheck removeSubscriber(ctx)
 
-  if ctx.state != Disabled:
+  if ctx.state != Disabled and ctx.beenConnected:
     # The clients `state` is set to `Disabled`, if we cannot accept their
     # `Connect`-packet and respond with a `ConnAck`. Since we dont accept
     # the connection, the client is never added to `mqttbroker.connections`.
@@ -54,24 +66,35 @@ proc processClient(s: AsyncSocket) {.async.} =
       if mqttbroker.retained[top].clientid == ctx.clientid:
         mqttbroker.retained.del(top)
 
-  if not ctx.s.isClosed():
+  if not ctx.s.isClosed() and ctx.beenConnected:
     ctx.s.close()
   ctx.state = Disabled
 
   if mqttbroker.verbosity >= 3:
     verbose(ctx)
 
 
-
 proc serve(host: string, port: int) {.async.} =
   var broker = newAsyncSocket()
   broker.setSockOpt(OptReuseAddr, true)
   broker.bindAddr(Port(port), host)
   broker.listen()
 
-  while true:
-    let client = await broker.accept()
-    asyncCheck processClient(client)
+  if mqttbroker.sslOn:
+    if not fileExists(mqttbroker.sslCert) or not fileExists(mqttbroker.sslKey):
+      echo "SSL cert or key does not exist. Check the path or generate them:\n" &
+           "openssl req -x509 -nodes -days 365 -newkey rsa:4096 -keyout mykey.pem -out mycert.pem"
+    var sslCtx = newContext(certFile = mqttbroker.sslCert, keyFile = mqttbroker.sslKey)
+    wrapSocket(sslCtx, broker)
+    while true:
+      let client = await broker.accept()
+      wrapConnectedSocket(sslCtx, client, handshakeAsServer)
+      asyncCheck processClient(client)
+
+  else:
+    while true:
+      let client = await broker.accept()
+      asyncCheck processClient(client)
 
 
 proc showConf(mb: MqttBroker, configfile: string) =
@@ -143,6 +166,12 @@ proc loadConf(mb: MqttBroker, config: string) =
   mqttbroker.passClientId     = parseBool(dict.getSectionValue("","clientid_pass"))
   mqttbroker.clientKickOld    = parseBool(dict.getSectionValue("","client_kickold"))
   mqttbroker.maxConnections   = parseInt(dict.getSectionValue("","max_conn"))
+  mqttbroker.sslCert          = dict.getSectionValue("","ssl_certificate")
+  mqttbroker.sslKey           = dict.getSectionValue("","ssl_key")
+
+  # If both certificate and key are present use SSL.
+  if mqttbroker.sslCert != "" and mqttbroker.sslKey != "":
+    mqttbroker.sslOn = true
 
   # If anonymous login is allowed return
   if not parseBool(dict.getSectionValue("","allow_anonymous")):
@@ -154,20 +183,18 @@ proc loadConf(mb: MqttBroker, config: string) =
 
 proc handler() {.noconv.} =
   ## Catch ctrl+c from user
+  echo " "
   if mqttbroker.verbosity >= 3:
-    echo " "
     verbose(mqttbroker)
-  echo "\nQuitting..\n"
   quit()
-setControlCHook(handler)
 
 
 proc nmqttBroker(config="", host="127.0.0.1", port=1883, verbosity=0, max_conn=0,
                   clientid_maxlen=60, clientid_spaces=false, clientid_empty=false,
-                  client_kickold=false, clientid_pass=false, password_file=""
+                  client_kickold=false, clientid_pass=false, password_file="",
+                  ssl=false, ssl_cert="", ssl_key=""
                 ) {.async.} =
   ## CLI tool for a MQTT broker
-
   if config != "":
     loadConf(mqttbroker, config)
   else:
@@ -181,30 +208,33 @@ proc nmqttBroker(config="", host="127.0.0.1", port=1883, verbosity=0, max_conn=0
     mqttbroker.clientKickOld    = client_kickold
     mqttbroker.passClientId     = clientid_pass
     mqttbroker.maxConnections   = max_conn
-    #mqttbroker.retainExpire = 3600
 
     if password_file != "":
       loadPasswords(password_file)
-      echo mqttbroker.passwords
 
-  #  if ssl:
-  #    newContext()
-  #    path to Key and Cert
-  #  if passwords:
-  #    password & username
+    if ssl:
+      mqttbroker.sslOn   = true
+      mqttbroker.sslCert = ssl_cert
+      mqttbroker.sslKey  = ssl_key
 
   if mqttbroker.verbosity >= 1:
     showConf(mqttbroker, config)
 
-  #let broker = newAsyncSocket()
-
   asyncCheck serve(host, port)
 
+  setControlCHook(handler)
+
+  runForever()
+
 
 
 when isMainModule:
 
-  let topLvlUse = """${doc}
+  let topLvlUse = """nmqtt version """ & nmqttVersion & """
+
+
+nmqtt is a MQTT v3.1.1 broker
+
 USAGE
   $command [options]
   $command [-c /path/to/config.conf]
@@ -225,7 +255,6 @@ $options
   clCfg.hTabCols = @[clOptKeys, clDescrip]
 
   dispatchGen(nmqttBroker,
-          doc="nmqtt is a MQTT v3.1.1 broker",
           cmdName="nmqtt",
           help={
             "config":           "absolute path to the config file. Overrides all other options.",
@@ -238,16 +267,20 @@ $options
             "clientid-empty":   "allow empty clientid and assign random id. Defaults to false.",
             "client-kickold":   "kick old client, if new client has same clientid. Defaults to false.",
             "clientid-pass":    "pass clientid in payload {clientid:payload}. Defaults to false.",
-            "password-file":    "absolute path to the password file"
+            "password-file":    "absolute path to the password file",
+            "ssl":              "activate ssl for the broker - requires --ssl-cert and --ssl-key.",
+            "ssl-cert":         "absolute path to the ssl certificate.",
+            "ssl-key":          "absolute path to the ssl key."
           },
           short={
             "help": '?',
-            "max-conn": '\0'
+            "max-conn": '\0',
+            "ssl": '\0',
+            "ssl-cert": '\0',
+            "ssl-key": '\0'
           },
           usage=topLvlUse,
           dispatchName="brokerCli"
           )
 
-  asyncCheck brokerCli(skipHelp=true)
-
-  runForever()
+  cligenQuit brokerCli(skipHelp=true)