view release on metacpan or  search on metacpan

src/EV__Nats.xs  view on Meta::CPAN

            char *e = line + len;
            char *found;

            /* max_payload accumulates into a signed int: parse wide and
               clamp to INT_MAX so a hostile INFO cannot poison it negative */
            found = (char *)nats_memfind(p, (size_t)(e - p), "\"max_payload\":", 14);
            if (found) {
                found += 14;
                uint64_t mp = 0;
                char *digits = found;
                while (found < e && *found >= '0' && *found <= '9') {
                    uint64_t d = (uint64_t)(*found - '0');
                    if (mp <= (UINT64_MAX - d) / 10)
                        mp = mp * 10 + d;
                    else
                        mp = UINT64_MAX;
                    found++;
                }
                if (found > digits && mp > 0)
                    self->max_payload = mp > (uint64_t)INT_MAX ? INT_MAX : (int)mp;
            }

            nats_parse_connect_urls(self, p, e - p);

            /* Parse nonce for NKey auth */
            found = (char *)nats_memfind(p, (size_t)(e - p), "\"nonce\":\"", 9);
            if (found) {
                found += 9;
                const char *nend = memchr(found, '"', e - found);
                if (nend) {
                    size_t nlen = nend - found;
                    if (self->server_nonce) Safefree(self->server_nonce);
                    Newx(self->server_nonce, nlen + 1, char);
                    memcpy(self->server_nonce, found, nlen);
                    self->server_nonce[nlen] = '\0';
                }
            }

            /* Parse ldm (lame duck mode) */
            found = (char *)nats_memfind(p, (size_t)(e - p), "\"ldm\":", 6);
            if (found) {
                found += 6;
                while (found < e && (*found == ' ' || *found == '\t')) found++;
                if (found < e && *found == 't') {
                    if (!self->ldm) {
                        self->ldm = 1;
                        if (self->on_ldm) {
                            dSP;
                            ENTER; SAVETMPS;
                            PUSHMARK(SP);
                            PUTBACK;
                            PINNED_CALL_SV(self->on_ldm, G_DISCARD);
                            FREETMPS; LEAVE;
                        }
                    }
                }
            }
        }

        if (self->connecting) {
            /* tls_required: CONNECT would leak credentials over plaintext
               before the server could reject it; refuse instead */
            {
                const char *tr = nats_memfind(line + 5, len - 5, "\"tls_required\":", 15);
                if (tr) {
                    const char *tre = line + len;
                    int tls_active = 0;
                    tr += 15;
                    while (tr < tre && (*tr == ' ' || *tr == '\t')) tr++;
#ifdef HAVE_OPENSSL
                    tls_active = self->tls;
#endif
                    if ((size_t)(tre - tr) >= 4 && memcmp(tr, "true", 4) == 0 && !tls_active) {
                        self->intentional_disconnect = 1;
                        nats_emit_error(self, "server requires TLS but TLS is not enabled");
                        nats_cleanup(self);
                        return;
                    }
                }
            }
#ifdef HAVE_OPENSSL
            if (self->tls && !self->ssl) {
                if (nats_ssl_setup(self) != 0) {
                    nats_ssl_fail(self, "SSL setup failed");
                    return;
                }
                self->ssl_handshaking = 1;
                int hret = nats_ssl_handshake(self);
                if (hret < 0) {
                    nats_ssl_fail(self, "SSL handshake failed");
                    return;
                }
                if (hret == 1) {
                    self->ssl_handshaking = 0;
                    nats_send_connect(self);
                    nats_try_write(self);
                }
                /* hret == 0: handshake in progress; nats_on_read will resume. */
                return;
            }
#endif
            nats_send_connect(self);
            nats_try_write(self);
        }
        return;
    }

    /* MSG */
    if (len >= 4 && line[0] == 'M' && line[1] == 'S' && line[2] == 'G' && line[3] == ' ') {
        if (nats_parse_msg_args(self, line, len) == 0) {
            /* max_payload is a signed int a hostile INFO could poison;
               the +2 check rejects lengths whose trailing CRLF would wrap */
            size_t limit = self->max_payload > 0 ? (size_t)self->max_payload
                                                 : (size_t)DEFAULT_MAX_PAYLOAD;
            if (self->msg_total_len > limit || self->msg_total_len > SIZE_MAX - 2) {
                nats_emit_error(self, "server sent message exceeding max_payload");
                nats_cleanup(self);
                return;
            }
            self->parse_state = PARSE_MSG_BODY;
        }