[MQTT / Mosca] Mqtt Packet 에 Command Type 추가하기 (for authentication) #2

너무 길어질 것 같아서 잘랐다.

[MQTT / Mosca] Mqtt Packet 에 Command Type 추가하기 (for authentication) #1

요기서 Connect 를 받지 않은 채로 Server - Device 간 통신을 어찌할지에 대해 언급만 하고 끝났다.

사실 더 많은 삽질을 했었는데, (Mosca, mqtt 라이브러리의 stream에 직접 접근한다거나..)

아무래도 라이브러리 내부적으로 처리하는 것들이 많다보니, 그렇게 직접 접근해서 데이터를 갖다 박아버리는 방식은 적합하지 않은 것 같았다.

가는 건 되는데 오는 게 안 된다거나, 보내고 받고 보냈는데 마지막 받는 걸 못한다거나..

그래서 라이브러리 자체 로직을 따르기로 했다.

Mqtt packet 을 generating, parsing 하는 부분 코드를 참조해서 Authentication 을 위한 Packet Type 을 만들어버리는 방법으로!

#MQTT Packet Command 추가하기

Authentication 을 위한 Packet 흐름이

1. Connect Packet (Device -> Server)

2. Authenticate Packet (Server -> Device)

3. Authenticate Packet (Device -> Server)

4. Connect ACK Packet (Server -> Device)

니까!!

2 단계인 Server -> Device 부터 보자.

#1. Mosca Authenticate

Mosca 에서 제공하는 Authentication 기능과 적절히 섞어서 사용할 거기 때문에,

Mosca 의 authenticate function 을 먼저 봐야 한다.

var authenticate = function(client, username, password, callback) {
  if (/* authentication success */)
    callback(null, true);
  else /* authentication failed */
    callback(null, false);
};

Mosca 의 기본 authentication 기능을 이용하려면, 위와 같이 authenticate function 을 내 입맛대로 정의해준 뒤에

Mosca Server setup 단계에서 아래와 같이 설정해주면 된다.

function setup() {
  server.authenticate = authenticate;
  /* authorizePublish, authorizeSubscribe 도 필요하면 셋팅해주면 된다. */
};

server.on('ready', setup);

당연한 얘기지만 setup function 을 server.on 에다 바로 정의해도 된다.

여튼 Mosca 의 Authentication 은 저렇게 쓰는 건데, 우린 저 단계에서 callback 으로 success parameter 를 넘기기 전에 device 와 통신을 해야한다.

그러기 위해서는 일단 Mosca 에서 Mqtt Packet 을 어떤 식으로 보내는지부터 봐야한다.

#2. Mosca authenticate function callback

Packet 전송을 어디서 어떻게 하는지 해당 코드를 찾는 가장 확실한 방법은

현재로선 위 authenticate function 에서 호출하도록 되어있는 callback 을 보는 것이다.

true, false 에 따라 Connect ACK Packet 을 보낼지 말지 결정할 거 아냐?

그래서 따라가봤다.

(mosca/lib/client.js)

that.server.authenticate(this, packet.username, packet.password,
                           function(err, verdict) {

    if (err) {
      logger.info({ username: packet.username }, "authentication error");
      client.connack({
        returnCode: 4
      });
      client.stream.end();
      return;
    }

    if (!verdict) {
      logger.info({ username: packet.username }, "authentication denied");
      client.connack({
        returnCode: 5
      });
      client.stream.end();
      return;
    }

    that.keepalive = packet.keepalive;
    that.will = packet.will;

    that.clean = packet.clean;

    if (that.id in that.server.clients){
      that.server.clients[that.id].close(completeConnection, "new connection request");
    } else {
      completeConnection();
    }
  });

요것이 해당코드!

저길 보면 client.connack(~) 하는 부분이 실제 Connect ACK Packet 을 전송하는 부분이라 볼 수 있다. (이름부터가.. ConnectACK -> CONNACK)

엥 근데 client? 

하고 client 가 뭔지 찾아올라가봤더니,

대체 mosca 에서 왜 이런 네이밍을 한 건지 모르겠지만, 

var that = this, logger, client = this.connection;

이란 부분이 있다.

그렇다면 저 connection 이란 애가 실제 Packet 전송을 담당한다는 건데.

쟤가 바로 'mqtt-connection' 이라는 module 이다.

#3. mqtt-connection

그럼 이제 이 모듈도 봐야 돼..

node_modules 에 있는 mqtt-connection directory 를 보면

mqtt-connection/connection.js 라는 파일이 있다.

걔를 열고 좀 내려보면

['connect',
  'connack',
  'publish',
  'puback',
  'pubrec',
  'pubrel',
  'pubcomp',
  'subscribe',
  'suback',
  'unsubscribe',
  'unsuback',
  'pingreq',
  'pingresp',
  'disconnect'].forEach(function(cmd) {
    Connection.prototype[cmd] = function(opts, cb) {
      opts = opts || {}
      opts.cmd = cmd;

      // flush the buffer if needed
      // UGLY hack, we should listen for the 'drain' event
      // and start writing again, but this works too
      this.write(opts)
      if (cb)
        setImmediate(cb)
    }
  });

이런 코드가 있는데,

저기 줄줄이 나열된 애들이 mqtt protocol 에서 원래 쓰는? command 들이다!

각 command 마다 

Connection.prototype.connect

뭐 이런 식으로 function 을 만들어주는 코드인데,

우리는 authenticate 를 쓸 거니까 저 배열에 은근슬쩍 'authenticate' 를 끼워주면 된다.

['connect',
  'connack',
  'publish',
  'puback',
  'pubrec',
  'pubrel',
  'pubcomp',
  'subscribe',
  'suback',
  'unsubscribe',
  'unsuback',
  'pingreq',
  'pingresp',
  'disconnect',
  'authenticate'].forEach(function(cmd) {
    Connection.prototype[cmd] = function(opts, cb) {
      opts = opts || {}
      opts.cmd = cmd;

      // flush the buffer if needed
      // UGLY hack, we should listen for the 'drain' event
      // and start writing again, but this works too
      this.write(opts)
      if (cb)
        setImmediate(cb)
    }
  });

이렇게!! 거의 다른그림찾기 수준,, ㅎㅎ

여튼 이렇게 해두고

지금은 [2. Authenticate Packet (Server -> Device)] 를 보고 있는 거니까!

서버에서 보내는 상황!

그럼 authenticate type 에 맞게 MQTT Packet 을 generate 해줘야겠지?

(mqtt-connection/lib/generateStream.js)

var through   = require('through2')
  , generate  = require('mqtt-packet').generate
  , empty     = new Buffer(0)

function generateStream() {
  var stream  = through.obj(process)

  function process(chunk, enc, cb) {
    var packet = empty;

    try {
      packet = generate(chunk)
    } catch(err) {
      this.emit('error', err)
      return;
    }

    this.push(packet)
    cb()
  }

  return stream
}

module.exports = generateStream;

응?

generateStream.js 를 보니,,

얘 정작 Packet generate 하는 데에는 또 다른 모듈을 쓴다!! ㅠㅠ

봐야할 것 ++..

그럼 이번엔 mqtt-packet 이란 모듈을 한번 보자.

#4. mqtt-packet (generate)

요 라이브러리도 코드를 열어보면, 딱 앗 얘가 generate 하는 애구나!! 싶은 파일이 있다. (이름이 generate.js 거든..)

근데 generate.js 를 보기 전에!!

파일 목록을 찬찬히 보다보면 거슬리는 놈이 하나 있다.

constants.js!!

이름부터가 뭔가 상수관리를 할 것 같은, 아주 중요해보이는 아이니까

얘를 먼저 한번 보자.

/* Command code => mnemonic */
module.exports.types = {
  0: 'reserved' /* -> authenticate */,
  1: 'connect',
  2: 'connack',
  3: 'publish',
  4: 'puback',
  5: 'pubrec',
  6: 'pubrel',
  7: 'pubcomp',
  8: 'subscribe',
  9: 'suback',
  10: 'unsubscribe',
  11: 'unsuback',
  12: 'pingreq',
  13: 'pingresp',
  14: 'disconnect',
  15: 'reserved'
};

호엑 역시 중요한 칭구였어

Packet Command 관련 상수들이 관리되는 파일인 것 같다.

1~14 까지는 다 원래 중요하게 쓰이는 Command 들이니 건드리지 말고

아래 위로 껴있는 reserved 중 하나를 authenticate 로 바꿔보자.

사실 reserved 가 무슨 역할인지는 아직 모른다.

0, 15 둘 중 아무거나 바꿔도 된다. 둘 다 테스트 해봤는데 잘 됨!

근데 신기한 건, 모든 과정을 끝내고 테스트하면서 Packet Capture를 떠보면, Authenticate Packet 도 Reserve 로 잡힌다.

MQTT Protocol 약속인가봐...

약속된 값을 건드렸는데도 문제가 없음에 감사하며 ㅠㅠ

constants.js 를 무사히 수정했다면 다시 원래 목적인 generate.js 를 보자.

(mqtt-packet/generate.js)

function generate(packet) {

  switch (packet.cmd) {
    case 'connect':
      return connect(packet)
    case 'authenticate':
      return authenticate(packet)
    case 'connack':
      return connack(packet)
    case 'publish':
      return publish(packet)
    case 'puback':
    case 'pubrec':
    case 'pubrel':
    case 'pubcomp':
    case 'unsuback':
      return confirmation(packet)
    case 'subscribe':
      return subscribe(packet)
    case 'suback':
      return suback(packet)
    case 'unsubscribe':
      return unsubscribe(packet)
    case 'pingreq':
    case 'pingresp':
    case 'disconnect':
      return emptyPacket(packet)
    default:
      throw new Error('unknown command')
  }
}

일단 파일을 열자마자 7번째 줄에 이런 코드가 있을 거다.

저기서 case 문들 중 아무 곳에나 'authenticate' case 를 끼워넣자.

나는 connect 다음에다 끼워넣었다.

근데 내가 적은 코드를 자세히 보니

return authenticate(packet)??

저게 원래 있던 function일까?

그럴리가.. 직접 추가해줘야한다.

이미 라이브러리에 포함되어 있는

다른 관련 function 코드를 참고해서 만들면 된다.

나는 username과 password를 포함한다는 점에서,

connect 와 비슷한 점이 많은 것 같아 connect function 을 복붙해서 고쳤다.

필요한 정보는 남기고, 불필요한 정보는 지우고,

필요한지 불필요한지 잘모르겠는 건 일단 남기고 ㅎㅎㅎㅎ

그렇게 짜깁기로 완성한 authenticate function code ▼

function authenticate(opts) {
  var opts = opts || {}
    , cmd = opts.cmd
    , protocolId = opts.protocolId || 'MQTT'
    , protocolVersion = opts.protocolVersion || 4
    , clean = opts.clean
    , keepalive = opts.keepalive || 0
    , clientId = opts.clientId || ""
    , username = opts.username
    , password = opts.password

  if (clean === undefined) {
    clean = true
  }

  var length = 0

  // Must be a string and non-falsy
  if (!protocolId ||
     (typeof protocolId !== "string" && !Buffer.isBuffer(protocolId))) {
    throw new Error('Invalid protocol id')
  } else {
    length += protocolId.length + 2
  }

  // Must be a 1 byte number
  if (!protocolVersion ||
      'number' !== typeof protocolVersion ||
      protocolVersion > 255 ||
      protocolVersion < 0) {

    throw new Error('Invalid protocol version')
  } else {
    length += 1
  }

  // ClientId might be omitted in 3.1.1, but only if cleanSession is set to 1
  if ((typeof clientId === "string" || Buffer.isBuffer(clientId)) &&
     (clientId || protocolVersion == 4) &&
     (clientId || clean)) {

    length += clientId.length + 2
  } else {

    if(protocolVersion < 4) {

      throw new Error('clientId must be supplied before 3.1.1');
    }

    if(clean == 0) {

      throw new Error('clientId must be given if cleanSession set to 0');
    }
  }

  // Must be a two byte number
  if ('number' !== typeof keepalive ||
      keepalive < 0 ||
      keepalive > 65535) {
    throw new Error('Invalid keepalive')
  } else {
    length += 2
  }

  // Connect flags
  length += 1

  // Username
  if (username) {
    if (username.length) {
      length += Buffer.byteLength(username) + 2
    } else {
      throw new Error('Invalid username')
    }
  }

  // Password
  if (password) {
    if (password.length) {
      length += byteLength(password) + 2
    } else {
      throw new Error('Invalid password')
    }
  }

  var buffer = new Buffer(1 + calcLengthLength(length) + length)
    , pos = 0

  // Generate header
  buffer.writeUInt8(protocol.codes[cmd] << protocol.CMD_SHIFT, pos++, true)

  // Generate length
  pos += writeLength(buffer, pos, length)

  // Generate protocol ID
  pos += writeStringOrBuffer(buffer, pos, protocolId)
  buffer.writeUInt8(protocolVersion, pos++, true)

  // Connect flags
  var flags = 0
  flags |= username ? protocol.USERNAME_MASK : 0
  flags |= password ? protocol.PASSWORD_MASK : 0
  flags |= clean ? protocol.CLEAN_SESSION_MASK : 0

  buffer.writeUInt8(flags, pos++, true)

  // Keepalive
  pos += writeNumber(buffer, pos, keepalive)

  // Client ID
  pos += writeStringOrBuffer(buffer, pos, clientId)

  // Username and password
  if (username)
    pos += writeStringOrBuffer(buffer, pos, username)

  if (password)
    pos += writeStringOrBuffer(buffer, pos, password)

  return buffer
}

이제 여기까지 했으면

authenticate type packet 을 generate 하는 것부터, 

mqtt-connection 라이브러리에 해당 타입(authenticate)의 패킷을 전송하는 function도 추가가 되었다.

이제 보내기만 하면 되지!!

var secretKey = +new Date();

client.connection.authenticate({
  clientId: client.id,
  username: "",
  password: secretKey.toString()
});

이게 실제로 보내는 코드다.

필요한대로 활용하면 된다.

나는 이 코드를 authenticate function 안에 넣어뒀다.

으으 이제 Device 에서 authenticate type Packet 을 받는 부분을 손봐야 한다.

원래 Server -> Device 부분을 한 글에 끝내버리려고 했는데...........

넘 길어져버려서 이쯤에서 끊어야 할 것 같다

여기서 더 쓰면 나중에 내가 다시 보기 힘들 듯 ㅎㅎ

안녕!!