diff options
| -rw-r--r-- | src/afb-hreq.h | 5 | ||||
| -rw-r--r-- | src/afb-websock.c | 20 | ||||
| -rw-r--r-- | src/afb-ws.c | 388 | ||||
| -rw-r--r-- | src/afb-ws.h | 20 | ||||
| -rw-r--r-- | src/websock.c | 72 | ||||
| -rw-r--r-- | src/websock.h | 23 |
6 files changed, 481 insertions, 47 deletions
diff --git a/src/afb-hreq.h b/src/afb-hreq.h index 8fbc5634..37bda99e 100644 --- a/src/afb-hreq.h +++ b/src/afb-hreq.h @@ -61,9 +61,4 @@ extern struct afb_req afb_hreq_to_req(struct afb_hreq *hreq); extern struct AFB_clientCtx *afb_hreq_context(struct afb_hreq *hreq); -extern int afb_hreq_one_page_api_redirect(struct afb_hreq *hreq, void *data); - -extern int afb_hreq_websocket_switch(struct afb_hreq *hreq, void *data); - -extern int afb_hreq_rest_api(struct afb_hreq *hreq, void *data); diff --git a/src/afb-websock.c b/src/afb-websock.c index 878df042..1f800717 100644 --- a/src/afb-websock.c +++ b/src/afb-websock.c @@ -170,7 +170,8 @@ static struct websock_itf aws_itf = { .on_close = (void*)aws_on_close, .on_text = (void*)aws_on_content, .on_binary = (void*)aws_on_content, - .on_continue = (void*)aws_on_content + .on_continue = (void*)aws_on_content, + .on_extension = NULL, }; struct afb_wsreq @@ -235,23 +236,25 @@ struct afb_websock *afb_websock_create(struct afb_hreq *hreq) result->tokener = json_tokener_new(); if (result->tokener == NULL) - goto error2; + goto error3; - result->ws = websock_create(&aws_itf, result); + result->ws = websock_create_v13(&aws_itf, result); if (result->ws == NULL) - goto error3; + goto error4; result->up = upoll_open(result->fd, result); if (result->up == NULL) - goto error4; + goto error5; upoll_on_readable(result->up, (void*)aws_on_readable); upoll_on_hangup(result->up, (void*)aws_disconnect); return result; -error4: +error5: websock_destroy(result->ws); -error3: +error4: json_tokener_free(result->tokener); +error3: + ctxClientPut(result->context); error2: free(result); error: @@ -301,7 +304,7 @@ static void aws_on_readable(struct afb_websock *ws) static int aws_handle_json(struct afb_websock *aws, struct json_object *obj) { struct afb_req r; - int count, num, rc; + int count, num; struct json_object *type, *id, *name, *req, *token; struct afb_wsreq *wsreq; const char *api, *verb; @@ -403,7 +406,6 @@ static void aws_on_content(struct afb_websock *ws, int last, size_t size) } } - static struct afb_arg wsreq_get(struct afb_wsreq *wsreq, const char *name) { struct afb_arg arg; diff --git a/src/afb-ws.c b/src/afb-ws.c new file mode 100644 index 00000000..e82e96fc --- /dev/null +++ b/src/afb-ws.c @@ -0,0 +1,388 @@ +/* + * Copyright 2016 IoT.bzh + * Author: José Bollo <jose.bollo@iot.bzh> + * + * Inspired by the work of + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#define _GNU_SOURCE +#include <assert.h> +#include <errno.h> +#include <sys/uio.h> +#include <string.h> + +#include <json.h> + +#include "websock.h" + +#include "utils-upoll.h" + +static ssize_t aws_writev(struct afb_ws *ws, const struct iovec *iov, int iovcnt); +static ssize_t aws_readv(struct afb_ws *ws, const struct iovec *iov, int iovcnt); +static void aws_disconnect(struct afb_ws *ws); +static void aws_on_close(struct afb_ws *ws, uint16_t code, size_t size); +static void aws_on_content(struct afb_ws *ws, int last, size_t size); +static void aws_on_readable(struct afb_ws *ws); + +static struct websock_itf aws_itf = { + .writev = (void*)aws_writev, + .readv = (void*)aws_readv, + .disconnect = (void*)aws_disconnect, + + .on_ping = NULL, + .on_pong = NULL, + .on_close = (void*)aws_on_close, + .on_text = (void*)aws_on_content, + .on_binary = (void*)aws_on_content, + .on_continue = (void*)aws_on_content, + .on_extension = NULL, +}; + +struct afb_wsreq +{ + struct afb_ws *aws; + struct afb_wsreq *next; + struct json_object *id; + struct json_object *name; + struct json_object *token; + struct json_object *request; +}; + +struct afb_ws +{ + int fd; + struct upoll *up; + struct websock *ws; + void (*cleanup)(void*); + void *cleanup_closure; + struct AFB_clientCtx *context; + struct afb_wsreq *requests; +}; + +static struct afb_arg wsreq_get(struct afb_wsreq *wsreq, const char *name); +static void wsreq_iterate(struct afb_wsreq *wsreq, int (*iterator)(void *closure, struct afb_arg arg), void *closure); +static void wsreq_fail(struct afb_wsreq *wsreq, const char *status, const char *info); +static void wsreq_success(struct afb_wsreq *wsreq, struct json_object *obj, const char *info); +static int wsreq_session_create(struct afb_wsreq *wsreq); +static int wsreq_session_check(struct afb_wsreq *wsreq, int refresh); +static void wsreq_session_close(struct afb_wsreq *wsreq); + +static const struct afb_req_itf wsreq_itf = { + .get = (void*)wsreq_get, + .iterate = (void*)wsreq_iterate, + .fail = (void*)wsreq_fail, + .success = (void*)wsreq_success, + .session_create = (void*)wsreq_session_create, + .session_check = (void*)wsreq_session_check, + .session_close = (void*)wsreq_session_close +}; + +struct afb_ws *afb_ws_create(int fd, struct AFB_clientCtx *context, void (*cleanup)(void*), void *closure) +{ + struct afb_ws *result; + + assert(fd >= 0); + assert(context != NULL); + + result = malloc(sizeof * result); + if (result == NULL) + goto error; + + result->fd = fd; + result->cleanup = cleanup; + result->cleanup_closure = closure; + result->context = ctxClientGet(context); + if (result->context == NULL) + goto error2; + + result->ws = websock_create_v13(&aws_itf, result); + if (result->ws == NULL) + goto error3; + + result->up = upoll_open(result->fd, result); + if (result->up == NULL) + goto error4; + + upoll_on_readable(result->up, (void*)aws_on_readable); + upoll_on_hangup(result->up, (void*)aws_disconnect); + return result; +error4: + websock_destroy(result->ws); +error3: + ctxClientPut(result->context); +error2: + free(result); +error: + close(fd); + return NULL; +} + +static ssize_t aws_writev(struct afb_ws *ws, const struct iovec *iov, int iovcnt) +{ + ssize_t rc; + do { + rc = writev(ws->fd, iov, iovcnt); + } while(rc == -1 && errno == EINTR); + return rc; +} + +static ssize_t aws_readv(struct afb_ws *ws, const struct iovec *iov, int iovcnt) +{ + ssize_t rc; + do { + rc = readv(ws->fd, iov, iovcnt); + } while(rc == -1 && errno == EINTR); + return rc; +} + +static void aws_disconnect(struct afb_ws *ws) +{ + upoll_close(ws->up); + websock_destroy(ws->ws); + close(ws->fd); + MHD_resume_connection (ws->connection); + ctxClientPut(ws->context); + json_tokener_free(ws->tokener); + free(ws); +} + +static void aws_on_close(struct afb_ws *ws, uint16_t code, size_t size) +{ + /* do nothing */ +} + +static void aws_on_readable(struct afb_ws *ws) +{ + websock_dispatch(ws->ws); +} + +static int aws_handle_json(struct afb_ws *aws, struct json_object *obj) +{ + struct afb_req r; + int count, num; + struct json_object *type, *id, *name, *req, *token; + struct afb_wsreq *wsreq; + const char *api, *verb; + size_t lenapi, lenverb; + + /* protocol inspired by http://www.gir.fr/ocppjs/ocpp_srpc_spec.shtml */ + + /* the object must be an array of 4 or 5 elements */ + if (!json_object_is_type(obj, json_type_array)) + goto error; + count = json_object_array_length(obj); + if (count < 4 || count > 5) + goto error; + + /* get the 5 elements: type id name request token */ + type = json_object_array_get_idx(obj, 0); + id = json_object_array_get_idx(obj, 1); + name = json_object_array_get_idx(obj, 2); + req = json_object_array_get_idx(obj, 3); + token = json_object_array_get_idx(obj, 4); + + /* check the types: int string string object string */ + if (!json_object_is_type(type, json_type_int)) + goto error; + if (!json_object_is_type(id, json_type_string)) + goto error; + if (!json_object_is_type(name, json_type_string)) + goto error; + if (!json_object_is_type(req, json_type_object)) + goto error; + if (token != NULL && !json_object_is_type(token, json_type_string)) + goto error; + + /* the type is only 2 */ + num = json_object_get_int(type); + if (num != 2) + goto error; + + /* checks the api/verb structure of name */ + api = json_object_get_string(name); + for (lenapi = 0 ; api[lenapi] && api[lenapi] != '/' ; lenapi++); + if (!lenapi || !api[lenapi]) + goto error; + verb = &api[lenapi+1]; + for (lenverb = 0 ; verb[lenverb] && verb[lenverb] != '/' ; lenverb++); + if (!lenverb || verb[lenverb]) + goto error; + + /* allocates the request data */ + wsreq = malloc(sizeof *wsreq); + if (wsreq == NULL) + goto error; + + /* fill and record the request */ + wsreq->aws = aws; + wsreq->id = json_object_get(id); + wsreq->name = json_object_get(name); + wsreq->token = json_object_get(token); + wsreq->request = json_object_get(req); + wsreq->next = aws->requests; + aws->requests = wsreq; + json_object_put(obj); + + r.data = wsreq; + r.itf = &wsreq_itf; + afb_apis_call(r, aws->context, api, lenapi, verb, lenverb); + return 1; + +error: + json_object_put(obj); + return 0; +} + +static void aws_on_content(struct afb_ws *ws, int last, size_t size) +{ + ssize_t rrc; + char buffer[8000]; + struct json_object *obj; + + json_tokener_reset(ws->tokener); + while(size) { + rrc = websock_read(ws->ws, buffer, + size > sizeof buffer ? sizeof buffer : size); + if (rrc < 0) { + websock_close(ws->ws); + return; + } + size -= (size_t)rrc; + obj = json_tokener_parse_ex(ws->tokener, buffer, (int)rrc); + if (obj != NULL) { + if (!aws_handle_json(ws, obj)) { + websock_close(ws->ws); + return; + } + } else if (json_tokener_get_error(ws->tokener) != json_tokener_continue) { + websock_close(ws->ws); + return; + } + } +} + +static struct afb_arg wsreq_get(struct afb_wsreq *wsreq, const char *name) +{ + struct afb_arg arg; + struct json_object *value; + + if (json_object_object_get_ex(wsreq->request, name, &value)) { + arg.name = name; + arg.value = json_object_get_string(value); + arg.size = strlen(arg.value); + } else { + arg.name = NULL; + arg.value = NULL; + arg.size = 0; + } + arg.path = NULL; + return arg; +} + +static void wsreq_iterate(struct afb_wsreq *wsreq, int (*iterator)(void *closure, struct afb_arg arg), void *closure) +{ + struct afb_arg arg; + struct json_object_iterator it = json_object_iter_begin(wsreq->request); + struct json_object_iterator end = json_object_iter_end(wsreq->request); + + arg.size = 0; + arg.path = NULL; + while(!json_object_iter_equal(&it, &end)) { + arg.name = json_object_iter_peek_name(&it); + arg.value = json_object_get_string(json_object_iter_peek_value(&it)); + if (!iterator(closure, arg)) + break; + json_object_iter_next(&it); + } +} + +static int wsreq_session_create(struct afb_wsreq *wsreq) +{ + struct AFB_clientCtx *context = wsreq->aws->context; + if (context->created) + return 0; + return wsreq_session_check(wsreq, 1); +} + +static int wsreq_session_check(struct afb_wsreq *wsreq, int refresh) +{ + const char *token; + struct AFB_clientCtx *context = wsreq->aws->context; + + if (wsreq->token == NULL) + return 0; + + token = json_object_get_string(wsreq->token); + if (token == NULL) + return 0; + + if (!ctxTokenCheck (context, token)) + return 0; + + if (refresh) { + ctxTokenNew (context); + } + + return 1; +} + +static void wsreq_session_close(struct afb_wsreq *wsreq) +{ + struct AFB_clientCtx *context = wsreq->aws->context; + ctxClientClose(context); +} + + +static void wsreq_reply(struct afb_wsreq *wsreq, int retcode, const char *status, const char *info, json_object *resp) +{ + json_object *root, *request, *reply; + const char *message; + + /* builds the answering structure */ + root = json_object_new_object(); + json_object_object_add(root, "jtype", json_object_new_string("afb-reply")); + request = json_object_new_object(); + json_object_object_add(root, "request", request); + json_object_object_add(request, "status", json_object_new_string(status)); + if (info) + json_object_object_add(request, "info", json_object_new_string(info)); + if (resp) + json_object_object_add(root, "response", resp); + + /* make the reply */ + reply = json_object_new_array(); + json_object_array_add(reply, json_object_new_int(retcode)); + json_object_array_add(reply, wsreq->id); + json_object_array_add(reply, root); + json_object_array_add(reply, json_object_new_string(wsreq->aws->context->token)); + + /* emits the reply */ + message = json_object_to_json_string(reply); + websock_text(wsreq->aws->ws, message, strlen(message)); + json_object_put(reply); + + /* TODO eliminates the wsreq */ +} + +static void wsreq_fail(struct afb_wsreq *wsreq, const char *status, const char *info) +{ + wsreq_reply(wsreq, 4, status, info, NULL); +} + +static void wsreq_success(struct afb_wsreq *wsreq, json_object *obj, const char *info) +{ + wsreq_reply(wsreq, 3, "success", info, obj); +} + diff --git a/src/afb-ws.h b/src/afb-ws.h new file mode 100644 index 00000000..6350fb9e --- /dev/null +++ b/src/afb-ws.h @@ -0,0 +1,20 @@ +/* + * Copyright 2016 IoT.bzh + * Author: José Bollo <jose.bollo@iot.bzh> + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +struct afb_ws; + + diff --git a/src/websock.c b/src/websock.c index 22e91be9..e583a49d 100644 --- a/src/websock.c +++ b/src/websock.c @@ -190,6 +190,22 @@ static int read_header(struct websock *ws) return 0; } +static int check_control_header(struct websock *ws) +{ + /* sanity checks */ + if (FRAME_GET_RSV1(ws->header[0]) != 0) + return 0; + if (FRAME_GET_RSV2(ws->header[0]) != 0) + return 0; + if (FRAME_GET_RSV3(ws->header[0]) != 0) + return 0; + if (FRAME_GET_MASK(ws->header[1])) + return 0; + if (FRAME_GET_OPCODE(ws->header[0]) == OPCODE_CLOSE) + return FRAME_GET_PAYLOAD_LEN(ws->header[1]) != 1; + return FRAME_GET_PAYLOAD_LEN(ws->header[1]) == 0; +} + int websock_dispatch(struct websock *ws) { loop: @@ -205,13 +221,6 @@ loop: return -1; else if (ws->lenhead < ws->szhead) return 0; - /* sanity checks */ - if (FRAME_GET_RSV1(ws->header[0]) != 0) - goto protocol_error; - if (FRAME_GET_RSV2(ws->header[0]) != 0) - goto protocol_error; - if (FRAME_GET_RSV3(ws->header[0]) != 0) - goto protocol_error; /* fast track */ switch (FRAME_GET_OPCODE(ws->header[0])) { case OPCODE_CONTINUATION: @@ -219,17 +228,13 @@ loop: case OPCODE_BINARY: break; case OPCODE_CLOSE: - if (FRAME_GET_MASK(ws->header[1])) - goto protocol_error; - if (FRAME_GET_PAYLOAD_LEN(ws->header[1]) == 1) + if (!check_control_header(ws)) goto protocol_error; if (FRAME_GET_PAYLOAD_LEN(ws->header[1])) ws->szhead += 2; break; case OPCODE_PING: - if (FRAME_GET_MASK(ws->header[1])) - goto protocol_error; - if (FRAME_GET_PAYLOAD_LEN(ws->header[1]) != 0) + if (!check_control_header(ws)) goto protocol_error; if (ws->itf->on_ping) ws->itf->on_ping(ws->closure); @@ -238,16 +243,14 @@ loop: ws->state = STATE_INIT; goto loop; case OPCODE_PONG: - if (FRAME_GET_MASK(ws->header[1])) - goto protocol_error; - if (FRAME_GET_PAYLOAD_LEN(ws->header[1]) != 0) + if (!check_control_header(ws)) goto protocol_error; if (ws->itf->on_pong) ws->itf->on_pong(ws->closure); ws->state = STATE_INIT; goto loop; default: - goto protocol_error; + break; } /* update heading size */ switch (FRAME_GET_PAYLOAD_LEN(ws->header[1])) { @@ -295,7 +298,30 @@ loop: ((unsigned char *)&ws->mask)[3] = ws->header[ws->szhead - 1]; } else ws->mask = 0; + + /* all heading fields are known, process */ ws->state = STATE_DATA; + if (ws->itf->on_extension != NULL) { + if (ws->itf->on_extension(ws->closure, + FRAME_GET_FIN(ws->header[0]), + FRAME_GET_RSV1(ws->header[0]), + FRAME_GET_RSV2(ws->header[0]), + FRAME_GET_RSV3(ws->header[0]), + FRAME_GET_OPCODE(ws->header[0]), + (size_t) ws->length)) { + return 0; + } + } + + /* not an extension case */ + if (FRAME_GET_RSV1(ws->header[0]) != 0) + goto protocol_error; + if (FRAME_GET_RSV2(ws->header[0]) != 0) + goto protocol_error; + if (FRAME_GET_RSV3(ws->header[0]) != 0) + goto protocol_error; + + /* handle */ switch (FRAME_GET_OPCODE(ws->header[0])) { case OPCODE_CONTINUATION: ws->itf->on_continue(ws->closure, @@ -320,9 +346,11 @@ loop: (size_t) ws->length); else ws->itf->on_close(ws->closure, - STATUS_CODE_UNSET, 0); + WEBSOCKET_CODE_UNSET, 0); ws->itf->disconnect(ws->closure); return 0; + default: + goto protocol_error; } break; @@ -338,11 +366,11 @@ loop: goto loop; too_long_error: - websock_close_code(ws, STATUS_CODE_MESSAGE_TOO_LARGE); + websock_close_code(ws, WEBSOCKET_CODE_MESSAGE_TOO_LARGE); return 0; protocol_error: - websock_close_code(ws, STATUS_CODE_PROTOCOL_ERROR); + websock_close_code(ws, WEBSOCKET_CODE_PROTOCOL_ERROR); return 0; } @@ -398,12 +426,12 @@ ssize_t websock_read(struct websock * ws, void *buffer, size_t size) void websock_drop(struct websock *ws) { - char buffer[4096]; + char buffer[8000]; while (ws->length && ws_read(ws, buffer, sizeof buffer) >= 0) ; } -struct websock *websock_create(const struct websock_itf *itf, void *closure) +struct websock *websock_create_v13(const struct websock_itf *itf, void *closure) { struct websock *result = calloc(1, sizeof *result); if (result) { diff --git a/src/websock.h b/src/websock.h index 235f9910..6376a37e 100644 --- a/src/websock.h +++ b/src/websock.h @@ -23,27 +23,28 @@ struct iovec; -#define STATUS_CODE_UNSET 0 -#define STATUS_CODE_OK 1000 -#define STATUS_CODE_GOING_AWAY 1001 -#define STATUS_CODE_PROTOCOL_ERROR 1002 -#define STATUS_CODE_RESERVED 1004 /* Protocol 8: frame too large */ -#define STATUS_CODE_INVALID_UTF8 1007 -#define STATUS_CODE_POLICY_VIOLATION 1008 -#define STATUS_CODE_MESSAGE_TOO_LARGE 1009 -#define STATUS_CODE_INTERNAL_ERROR 1011 +#define WEBSOCKET_CODE_UNSET 0 +#define WEBSOCKET_CODE_OK 1000 +#define WEBSOCKET_CODE_GOING_AWAY 1001 +#define WEBSOCKET_CODE_PROTOCOL_ERROR 1002 +#define WEBSOCKET_CODE_RESERVED 1004 /* Protocol 8: frame too large */ +#define WEBSOCKET_CODE_INVALID_UTF8 1007 +#define WEBSOCKET_CODE_POLICY_VIOLATION 1008 +#define WEBSOCKET_CODE_MESSAGE_TOO_LARGE 1009 +#define WEBSOCKET_CODE_INTERNAL_ERROR 1011 struct websock_itf { ssize_t (*writev) (void *, const struct iovec *, int); ssize_t (*readv) (void *, const struct iovec *, int); void (*disconnect) (void *); - void (*on_ping) (void *); + void (*on_ping) (void *); /* if not NULL, responsible of pong */ void (*on_pong) (void *); void (*on_close) (void *, uint16_t code, size_t size); void (*on_text) (void *, int last, size_t size); void (*on_binary) (void *, int last, size_t size); void (*on_continue) (void *, int last, size_t size); + int (*on_extension) (void *, int last, int rsv1, int rsv2, int rsv3, int opcode, size_t size); }; struct websock; @@ -61,6 +62,6 @@ void websock_drop(struct websock *ws); int websock_dispatch(struct websock *ws); -struct websock *websock_create(const struct websock_itf *itf, void *closure); +struct websock *websock_create_v13(const struct websock_itf *itf, void *closure); void websock_destroy(struct websock *ws); |