From c57fab661cfcbbdce75620a4a86a61bd41918a6d Mon Sep 17 00:00:00 2001 From: Granyaa Date: Thu, 15 Jan 2026 11:19:54 +0200 Subject: [PATCH] Modified Kong Plugin to support the new Nano Attachment --- .../open-appsec-waf-kong-plugin/handler.lua | 548 +++++++++++------- .../lua_attachment_wrapper.c | 197 ++++++- .../open-appsec-waf-kong-plugin/nano_ffi.lua | 128 +++- ...en-appsec-waf-kong-plugin-1.0.0-1.rockspec | 12 +- .../open-appsec-waf-kong-plugin/schema.lua | 15 +- 5 files changed, 633 insertions(+), 267 deletions(-) diff --git a/attachments/kong/plugins/open-appsec-waf-kong-plugin/handler.lua b/attachments/kong/plugins/open-appsec-waf-kong-plugin/handler.lua index 9836521..3f31f17 100755 --- a/attachments/kong/plugins/open-appsec-waf-kong-plugin/handler.lua +++ b/attachments/kong/plugins/open-appsec-waf-kong-plugin/handler.lua @@ -1,6 +1,7 @@ local module_name = ... local prefix = module_name:match("^(.-)handler$") local nano = require(prefix .. "nano_ffi") +local semaphore = require "ngx.semaphore" local kong = kong local NanoHandler = {} @@ -8,266 +9,399 @@ local NanoHandler = {} NanoHandler.PRIORITY = 3000 NanoHandler.VERSION = "1.0.0" -function NanoHandler.init_worker() - nano.init_attachment() +NanoHandler.sessions = {} +NanoHandler.processed_requests = {} + +-- per-worker state +local pending = {} -- sid -> { semaphore, verdict } + +local function drain_queue() + kong.log.debug("drain_queue: Starting to drain queue") + local drained_count = 0 + while not nano.is_queue_empty() do + local session_id = nano.pop_from_queue() + kong.log.debug("drain_queue: Popped session_id=", session_id or "nil") + if session_id and session_id > 0 then + local session_info = pending[session_id] + if session_info and session_info.sem then + kong.log.debug("drain_queue: Notifying semaphore for session_id=", session_id) + session_info.sem:post() + drained_count = drained_count + 1 + else + kong.log.warn("drain_queue: No semaphore found for session_id=", session_id) + end + else + kong.log.err("drain_queue: Invalid session_id=", session_id or "nil") + end + end + kong.log.debug("drain_queue: Drained ", drained_count, " sessions") end +local function start_verdict_listener() + local socket_fd = nano.get_attachment_socket() + if not socket_fd or socket_fd < 0 then + kong.log.err("Failed to get attachment socket") + return + end + + kong.log.info("Starting verdict listener on socket fd: ", socket_fd) + + ngx.timer.at(0, function(premature) + if premature then + return + end + + local sock = ngx.socket.tcp() + sock:settimeout(1000) -- 1 second timeout + + -- Set the socket to the existing file descriptor + local ok, err = sock:setfd(socket_fd) + if not ok then + kong.log.err("Failed to set socket fd: ", err) + return + end + + kong.log.info("Listening on verdict socket") + + while true do + -- Use socket as a doorbell - wait for any data + local data, err, partial = sock:receive(1) + if not data and not partial then + if err == "timeout" then + -- Continue waiting + goto continue + else + kong.log.err("verdict_listener: Error receiving from verdict socket: ", err) + goto continue + end + end + + -- Socket has data - drain the queue + kong.log.debug("verdict_listener: Received doorbell signal, draining queue") + local ok, drain_err = pcall(drain_queue) + if not ok then + kong.log.err("verdict_listener: Error draining queue: ", drain_err) + end + + ::continue:: + end + end) +end + +function NanoHandler.init_worker() + nano.init_attachment() + start_verdict_listener() +end + +-- **Handles Request Headers (DecodeHeaders Equivalent)** function NanoHandler.access(conf) - local ctx = kong.ctx.plugin - + kong.log.debug("access: Starting access phase") local headers = kong.request.get_headers() local session_id = nano.generate_session_id() - + kong.log.debug("access: Generated session_id=", session_id) + kong.service.request.set_header("x-session-id", tostring(session_id)) + + if NanoHandler.processed_requests[session_id] then + kong.log.warn("access: Session already processed, blocking session_id=", session_id) + kong.ctx.plugin.blocked = true + return + end + local session_data = nano.init_session(session_id) if not session_data then - kong.ctx.plugin.cleanup_needed = false - return - end - - ctx.session_data = session_data - ctx.session_id = session_id - if nano.is_session_finalized(session_id) then - kong.log.debug("Session has already been inspected, no need for further inspection") + kong.log.err("access: Failed to initialize session for session_id=", session_id, " - failing open") return end + kong.log.debug("access: Session initialized successfully for session_id=", session_id) + + kong.ctx.plugin.session_data = session_data + kong.ctx.plugin.session_id = session_id local meta_data = nano.handle_start_transaction() if not meta_data then - kong.log.debug("Failed to handle start transaction - failing mode") - ctx.cleanup_needed = true + kong.log.err("access: Failed to handle start transaction for session_id=", session_id, " - failing open") return end - + kong.log.debug("access: Start transaction handled for session_id=", session_id) + local req_headers = nano.handleHeaders(headers) - if not req_headers then - kong.log.debug("Failed to handle request headers - failing mode") - ctx.cleanup_needed = true - return - end local has_content_length = tonumber(ngx.var.http_content_length) and tonumber(ngx.var.http_content_length) > 0 local contains_body = has_content_length and 1 or 0 - local verdict, response = nano.send_data(session_id, session_data, meta_data, req_headers, contains_body, nano.HttpChunkType.HTTP_REQUEST_FILTER) - if verdict ~= nano.AttachmentVerdict.INSPECT then - ctx.cleanup_needed = true - if verdict == nano.AttachmentVerdict.DROP then - return nano.handle_custom_response(session_data, response) - end + local sem = semaphore.new() + pending[session_id] = { sem = sem, verdict = nil } + + -- Use non-blocking send_data_async + kong.log.debug("access: Sending headers async for session_id=", session_id, " contains_body=", contains_body) + nano.send_data_async(session_id, session_data, meta_data, req_headers, contains_body, nano.HttpChunkType.HTTP_REQUEST_FILTER) + + -- Wait on semaphore for verdict + -- TODO get timeout from conf + kong.log.debug("access: Waiting for headers verdict for session_id=", session_id) + local ok, err = sem:wait(1) + + if not ok then + kong.log.err( + "access: Timeout while waiting for headers verdict for session_id=", session_id, " err=", err or "nil" + ) + -- Todo check if fail open/close + nano.fini_session(session_data) + nano.cleanup_all() return end + -- Query verdict after semaphore wakeup + kong.log.debug("access: Querying headers verdict for session_id=", session_id) + local verdict, response = nano.get_attachment_verdict_response(session_id) + kong.log.debug("access: Headers verdict=", verdict, " for session_id=", session_id) + + if verdict == nano.AttachmentVerdict.DROP then + kong.log.warn("access: Headers verdict DROP for session_id=", session_id) + nano.fini_session(session_data) + kong.ctx.plugin.blocked = true + local result = nano.handle_custom_response(session_data, response) + nano.cleanup_all() + return result + end + kong.log.debug("access: Headers verdict ACCEPT for session_id=", session_id) + if contains_body == 1 then + kong.log.debug("access: Request contains body for session_id=", session_id) local body = kong.request.get_raw_body() if body and #body > 0 then - verdict, response = nano.send_body(session_id, session_data, body, nano.HttpChunkType.HTTP_REQUEST_BODY) - if verdict ~= nano.AttachmentVerdict.INSPECT then - ctx.cleanup_needed = true - if verdict == nano.AttachmentVerdict.DROP then - return nano.handle_custom_response(session_data, response) - end + -- Use non-blocking send_body_async + kong.log.debug("access: Sending body async, size=", #body, " for session_id=", session_id) + nano.send_body_async(session_id, session_data, body, nano.HttpChunkType.HTTP_REQUEST_BODY) + + -- Wait on semaphore for verdict + kong.log.debug("access: Waiting for body verdict for session_id=", session_id) + local ok, err = sem:wait(1) + if not ok then + kong.log.err( + "access: Timeout while waiting for body verdict for session_id=", session_id, " err=", err or "nil" + ) + -- Todo check if fail open/close + nano.fini_session(session_data) + nano.cleanup_all() return end - else - local body_file = ngx.var.request_body_file - if body_file then - local file = io.open(body_file, "rb") - if file then - local chunk_size = 8192 - local chunk_count = 0 - local start_time = ngx.now() - local timeout_sec = nano.get_request_processing_timeout_sec() - kong.log.debug("Request body reading timeout set to ", timeout_sec, " seconds") - - while true do - ngx.update_time() - local current_time = ngx.now() - local elapsed = current_time - start_time - - if elapsed > timeout_sec then - ctx.cleanup_needed = true - kong.log.warn("Request body reading timeout after ", elapsed, " seconds") - file:close() - return - end - - local chunk = file:read(chunk_size) - if not chunk or #chunk == 0 then - kong.log.debug("End of request body file reached") - break - end - - chunk_count = chunk_count + 1 - kong.log.debug("Sending request body chunk ", chunk_count, " of size ", #chunk, " bytes to C module") - verdict, response = nano.send_body(session_id, session_data, chunk, nano.HttpChunkType.HTTP_REQUEST_BODY) - - if verdict ~= nano.AttachmentVerdict.INSPECT then - file:close() - ctx.cleanup_needed = true - if verdict == nano.AttachmentVerdict.DROP then - return nano.handle_custom_response(session_data, response) - end - return - end - end - file:close() - kong.log.debug("Sent ", chunk_count, " chunks from request body file") + + -- Query verdict after semaphore wakeup + kong.log.debug("access: Querying body verdict for session_id=", session_id) + local verdict, response = nano.get_attachment_verdict_response(session_id) + kong.log.debug("access: Body verdict=", verdict, " for session_id=", session_id) + if verdict == nano.AttachmentVerdict.DROP then + kong.log.warn("access: Body verdict DROP for session_id=", session_id) + nano.fini_session(session_data) + kong.ctx.plugin.blocked = true + local result = nano.handle_custom_response(session_data, response) + nano.cleanup_all() + return result + end + else + kong.log.debug("access: Request body not in memory, attempting to read from buffer/file for session_id=", session_id) + + local body_data = ngx.var.request_body + if body_data and #body_data > 0 then + kong.log.debug("access: Found request body in nginx var, size=", #body_data, " for session_id=", session_id) + verdict, response = nano.send_body(session_id, session_data, body_data, nano.HttpChunkType.HTTP_REQUEST_BODY) + kong.log.debug("access: Nginx var body verdict=", verdict, " for session_id=", session_id) + if verdict == nano.AttachmentVerdict.DROP then + kong.log.warn("access: Nginx var body verdict DROP for session_id=", session_id) + nano.fini_session(session_data) + kong.ctx.plugin.blocked = true + return nano.handle_custom_response(session_data, response) end else - kong.log.err("Request body expected but no body data or file available") + local body_file = ngx.var.request_body_file + if body_file then + kong.log.debug("access: Reading request body from file=", body_file, " for session_id=", session_id) + local file, open_err = io.open(body_file, "rb") + if file then + local entire_body = file:read("*all") + file:close() + + if entire_body and #entire_body > 0 then + kong.log.debug("access: Sending entire body from file, size=", #entire_body, " bytes for session_id=", session_id) + verdict, response = nano.send_body(session_id, session_data, entire_body, nano.HttpChunkType.HTTP_REQUEST_BODY) + kong.log.debug("access: File body verdict=", verdict, " for session_id=", session_id) + if verdict == nano.AttachmentVerdict.DROP then + kong.log.warn("access: File body verdict DROP for session_id=", session_id) + nano.fini_session(session_data) + kong.ctx.plugin.blocked = true + local result = nano.handle_custom_response(session_data, response) + nano.cleanup_all() + return result + end + else + kong.log.debug("access: Empty body file for session_id=", session_id) + end + else + kong.log.err("access: Failed to open body file=", body_file, " err=", open_err or "nil", " for session_id=", session_id) + end + else + kong.log.warn("access: Request body expected but no body data or file available for session_id=", session_id) + end end end - local ok, verdict, response = pcall(function() - return nano.end_inspection(session_id, session_data, nano.HttpChunkType.HTTP_REQUEST_END) + kong.log.debug("access: Ending request inspection (with body) for session_id=", session_id) + local ok, result = pcall(function() + return nano.end_inspection_async(session_id, session_data, nano.HttpChunkType.HTTP_REQUEST_END) end) if not ok then - kong.log.debug("Error ending request inspection: ", verdict, " - failing open") - ctx.cleanup_needed = true + kong.log.err("access: Error ending request inspection for session_id=", session_id, " err=", result, " - failing open") + nano.fini_session(session_data) + nano.cleanup_all() return end - if verdict ~= nano.AttachmentVerdict.INSPECT then - ctx.cleanup_needed = true - if verdict == nano.AttachmentVerdict.DROP then - return nano.handle_custom_response(session_data, response) - end + -- Wait on semaphore for verdict + kong.log.debug("access: Waiting for end inspection verdict for session_id=", session_id) + local ok, err = sem:wait(1) + if not ok then + kong.log.err("access: Timeout waiting for end inspection verdict for session_id=", session_id, " err=", err or "nil") + nano.fini_session(session_data) + nano.cleanup_all() return end - end -end -function NanoHandler.header_filter(conf) - local ctx = kong.ctx.plugin - if nano.is_session_finalized(ctx.session_data) then - kong.log.debug("Session has already been inspected, no need for further inspection") - return - end + -- Query verdict after semaphore wakeup + kong.log.debug("access: Querying end inspection verdict for session_id=", session_id) + local verdict, response = nano.get_attachment_verdict_response(session_id) + kong.log.debug("access: End inspection verdict=", verdict, " for session_id=", session_id) - if ctx.cleanup_needed then - kong.log.debug("cleanup in header_filter, passing through") - return - end - - local session_id = ctx.session_id - local session_data = ctx.session_data - - local headers = kong.response.get_headers() - local header_data = nano.handleHeaders(headers) - - if not header_data then - kong.log.debug("Failed to handle response headers - failing open") - ctx.cleanup_needed = true - return - end - - local status_code = kong.response.get_status() - local content_length = tonumber(headers["content-length"]) or 0 - - local verdict, response = nano.send_response_headers(session_id, session_data, header_data, status_code, content_length) - if verdict ~= nano.AttachmentVerdict.INSPECT then - ctx.cleanup_needed = true if verdict == nano.AttachmentVerdict.DROP then - kong.log.debug("DROP verdict in header_filter - sending block response immediately") - return nano.handle_custom_response(session_data, response) + kong.log.warn("access: End inspection verdict DROP for session_id=", session_id) + nano.fini_session(session_data) + kong.ctx.plugin.blocked = true + local result = nano.handle_custom_response(session_data, response) + nano.cleanup_all() + return result + end + else + kong.log.debug("access: Ending request inspection (no body) for session_id=", session_id) + nano.end_inspection_async(session_id, session_data, nano.HttpChunkType.HTTP_REQUEST_END) + + -- Wait on semaphore for verdict + kong.log.debug("access: Waiting for end inspection verdict (no body) for session_id=", session_id) + local ok, err = sem:wait(1) + if not ok then + kong.log.err("access: Timeout waiting for end inspection verdict (no body) for session_id=", session_id, " err=", err or "nil") + nano.fini_session(session_data) + nano.cleanup_all() + return + end + + -- Query verdict after semaphore wakeup + kong.log.debug("access: Querying end inspection verdict (no body) for session_id=", session_id) + local verdict, response = nano.get_attachment_verdict_response(session_id) + kong.log.debug("access: End inspection verdict (no body)=", verdict, " for session_id=", session_id) + + if verdict == nano.AttachmentVerdict.DROP then + kong.log.warn("access: End inspection verdict DROP (no body) for session_id=", session_id) + nano.fini_session(session_data) + kong.ctx.plugin.blocked = true + local result = nano.handle_custom_response(session_data, response) + nano.cleanup_all() + return result end - ngx.header["Content-Length"] = nil - return end - ngx.header["Content-Length"] = nil - - ctx.expect_body = not (status_code == 204 or status_code == 304 or (100 <= status_code and status_code < 200) or content_length == 0) + kong.log.debug("access: Request processing complete for session_id=", session_id) + pending[session_id] = nil + NanoHandler.processed_requests[session_id] = true end -function NanoHandler.body_filter(conf) - local ctx = kong.ctx.plugin - local chunk = ngx.arg[1] - local eof = ngx.arg[2] - - local session_id = ctx.session_id - local session_data = ctx.session_data - - if nano.is_session_finalized(session_data) then - kong.log.debug("Session has already been inspected, no need for further inspection") - return - end - - if ctx.cleanup_needed then - kong.log.debug("cleanup chunk without inspection, passing through") - return - end +-- function NanoHandler.header_filter(conf) +-- local ctx = kong.ctx.plugin +-- if ctx.blocked then +-- return +-- end - if not ctx.body_filter_start_time then - ctx.body_filter_start_time = ngx.now() - ctx.body_filter_timeout_sec = nano.get_response_processing_timeout_sec() - kong.log.debug("body_filter timeout set to ", ctx.body_filter_timeout_sec, " seconds") - end - - local elapsed_time = ngx.now() - ctx.body_filter_start_time - if elapsed_time > ctx.body_filter_timeout_sec then - kong.log.warn("Body filter timeout after ", elapsed_time, " seconds - failing open") - ctx.cleanup_needed = true - return - end +-- local session_id = ctx.session_id +-- local session_data = ctx.session_data - if chunk and #chunk > 0 then - ctx.body_buffer_chunk = ctx.body_buffer_chunk or 0 - ctx.body_seen = true +-- if not session_id or not session_data then +-- return +-- end - local verdict, response, modifications = nano.send_body(session_id, session_data, chunk, nano.HttpChunkType.HTTP_RESPONSE_BODY) - - if modifications then - chunk = nano.handle_body_modifications(chunk, modifications, ctx.body_buffer_chunk) - end +-- local headers = kong.response.get_headers() +-- local header_data = nano.handleHeaders(headers) +-- local status_code = kong.response.get_status() +-- local content_length = tonumber(headers["content-length"]) or 0 - ctx.body_buffer_chunk = ctx.body_buffer_chunk + 1 +-- local verdict, response = nano.send_response_headers(session_id, session_data, header_data, status_code, content_length) +-- if verdict == nano.AttachmentVerdict.DROP then +-- kong.ctx.plugin.blocked = true +-- nano.fini_session(session_data) +-- nano.cleanup_all() +-- return nano.handle_custom_response(session_data, response) +-- end - if verdict ~= nano.AttachmentVerdict.INSPECT then - ctx.cleanup_needed = true - if verdict == nano.AttachmentVerdict.DROP then - kong.log.debug("DROP verdict during response streaming - closing connection") - ngx.header["Connection"] = "close" - ngx.arg[1] = "" - ngx.arg[2] = true - return - end - end - - ngx.arg[1] = chunk - return - end +-- ctx.expect_body = not (status_code == 204 or status_code == 304 or (100 <= status_code and status_code < 200) or content_length == 0) +-- end - if eof then - if ctx.body_seen or ctx.expect_body == false then - ctx.cleanup_needed = true - local verdict, response = nano.end_inspection(session_id, session_data, nano.HttpChunkType.HTTP_RESPONSE_END) - if verdict ~= nano.AttachmentVerdict.INSPECT then - kong.log.debug("Final verdict after end_inspection: ", verdict) - ctx.cleanup_needed = true - if verdict == nano.AttachmentVerdict.DROP then - kong.log.debug("DROP verdict at EOF - closing connection") - ngx.header["Connection"] = "close" - ngx.arg[1] = "" - ngx.arg[2] = true - return - end - end - end - - end -end +-- function NanoHandler.body_filter(conf) +-- local ctx = kong.ctx.plugin +-- if ctx.blocked then +-- return +-- end -function NanoHandler.log(conf) - local ctx = kong.ctx.plugin - if ctx.cleanup_needed then - nano.fini_session(ctx.session_data) - nano.cleanup_all() - ctx.session_data = nil - ctx.session_id = nil - collectgarbage("collect") - end -end +-- local session_id = ctx.session_id +-- local session_data = ctx.session_data + +-- if not session_id or not session_data or ctx.session_finalized then +-- return +-- end + +-- local body = kong.response.get_raw_body() + +-- if body then +-- ctx.body_seen = true +-- local verdict, response, modifications = nano.send_body(session_id, session_data, body, nano.HttpChunkType.HTTP_RESPONSE_BODY) + +-- -- Initialize if not exists +-- ctx.body_buffer_chunk = ctx.body_buffer_chunk or 0 + +-- -- Handle body modifications if any +-- if modifications then +-- body = nano.handle_body_modifications(body, modifications, ctx.body_buffer_chunk) +-- kong.response.set_raw_body(body) +-- end + +-- ctx.body_buffer_chunk = ctx.body_buffer_chunk + 1 + +-- if verdict == nano.AttachmentVerdict.DROP then +-- nano.fini_session(session_data) +-- ctx.session_finalized = true +-- local result = nano.handle_custom_response(session_data, response) +-- -- Clean up allocated memory +-- nano.cleanup_all() +-- return result +-- end +-- return +-- end + +-- if ctx.body_seen or ctx.expect_body == false then +-- local verdict, response = nano.end_inspection(session_id, session_data, nano.HttpChunkType.HTTP_RESPONSE_END) +-- if verdict == nano.AttachmentVerdict.DROP then +-- nano.fini_session(session_data) +-- ctx.session_finalized = true +-- local result = nano.handle_custom_response(session_data, response) +-- -- Clean up allocated memory +-- nano.cleanup_all() +-- return result +-- end + +-- nano.fini_session(session_data) +-- -- Clean up allocated memory +-- nano.cleanup_all() +-- ctx.session_finalized = true +-- end +-- end + +return NanoHandler -return NanoHandler \ No newline at end of file diff --git a/attachments/kong/plugins/open-appsec-waf-kong-plugin/lua_attachment_wrapper.c b/attachments/kong/plugins/open-appsec-waf-kong-plugin/lua_attachment_wrapper.c index eb49244..bdbe530 100755 --- a/attachments/kong/plugins/open-appsec-waf-kong-plugin/lua_attachment_wrapper.c +++ b/attachments/kong/plugins/open-appsec-waf-kong-plugin/lua_attachment_wrapper.c @@ -12,7 +12,7 @@ static int lua_init_nano_attachment(lua_State *L) { int worker_id = luaL_checkinteger(L, 1); int num_workers = luaL_checkinteger(L, 2); - NanoAttachment* attachment = InitNanoAttachment(0, worker_id, num_workers, fileno(stderr)); + NanoAttachment* attachment = InitNanoAttachment(0, worker_id, num_workers, fileno(stdout)); if (!attachment) { lua_pushnil(L); lua_pushstring(L, "Failed to initialize NanoAttachment"); @@ -135,7 +135,7 @@ static int lua_createNanoStrAlloc(lua_State *L) { lua_pushnil(L); lua_pushstring(L, "Failed to allocate memory for string"); return 2; - } +} nano_str_t* nanoStr = (nano_str_t*)malloc(sizeof(nano_str_t)); if (!nanoStr) { @@ -350,6 +350,43 @@ static int lua_send_data(lua_State *L) { return 2; } +static int lua_send_data_async(lua_State *L) { + NanoAttachment* attachment = (NanoAttachment*) lua_touserdata(L, 1); + SessionID session_id = luaL_checkinteger(L, 2); + HttpSessionData *session_data = (HttpSessionData*) lua_touserdata(L, 3); + HttpChunkType chunk_type = luaL_checkinteger(L, 4); + HttpMetaData* meta_data = (HttpMetaData*) lua_touserdata(L, 5); + HttpHeaders* req_headers = (HttpHeaders*) lua_touserdata(L, 6); + int contains_body = luaL_checkinteger(L, 7); + + if (!attachment || !session_data || !meta_data || !req_headers) { + lua_pushstring(L, "Error: received NULL data in lua_send_data_async"); + return lua_error(L); + } + + HttpRequestFilterData *filter_data = (HttpRequestFilterData *)malloc(sizeof(HttpRequestFilterData)); + if (!filter_data) { + return luaL_error(L, "Memory allocation failed for HttpRequestFilterData"); + } + + filter_data->meta_data = meta_data; + filter_data->req_headers = req_headers; + filter_data->contains_body = contains_body; + + AttachmentData attachment_data; + attachment_data.session_id = session_id; + attachment_data.session_data = session_data; + attachment_data.chunk_type = chunk_type; + attachment_data.data = (void*)filter_data; + + SendDataNanoAttachmentAsync(attachment, &attachment_data); + + free(filter_data); + + lua_pushinteger(L, session_id); + return 1; +} + static int lua_send_body(lua_State *L) { NanoAttachment* attachment = (NanoAttachment*) lua_touserdata(L, 1); SessionID session_id = luaL_checkinteger(L, 2); @@ -364,7 +401,7 @@ static int lua_send_body(lua_State *L) { } if (body_len <= 8 * 1024) { - HttpBody http_chunks; + NanoHttpBody http_chunks; http_chunks.bodies_count = 1; nano_str_t chunk; @@ -400,7 +437,7 @@ static int lua_send_body(lua_State *L) { num_chunks = 10000; } - HttpBody http_chunks; + NanoHttpBody http_chunks; http_chunks.bodies_count = num_chunks; http_chunks.data = (nano_str_t*)malloc(num_chunks * sizeof(nano_str_t)); @@ -441,6 +478,79 @@ static int lua_send_body(lua_State *L) { return 3; } +static int lua_send_body_async(lua_State *L) { + NanoAttachment* attachment = (NanoAttachment*) lua_touserdata(L, 1); + SessionID session_id = luaL_checkinteger(L, 2); + HttpSessionData *session_data = (HttpSessionData*) lua_touserdata(L, 3); + size_t body_len; + const char *body_chunk = luaL_checklstring(L, 4, &body_len); + HttpChunkType chunk_type = luaL_checkinteger(L, 5); + + if (!attachment || !session_data || !body_chunk) { + lua_pushstring(L, "Error: Invalid attachment or session_data"); + return lua_error(L); + } + + if (body_len <= 8 * 1024) { + NanoHttpBody http_chunks; + http_chunks.bodies_count = 1; + + nano_str_t chunk; + chunk.data = (unsigned char*)body_chunk; + chunk.len = body_len; + http_chunks.data = &chunk; + + AttachmentData attachment_data; + attachment_data.session_id = session_id; + attachment_data.session_data = session_data; + attachment_data.chunk_type = chunk_type; + attachment_data.data = &http_chunks; + + SendDataNanoAttachmentAsync(attachment, &attachment_data); + + lua_pushinteger(L, session_id); + return 1; + } + + const size_t CHUNK_SIZE = 8 * 1024; + size_t num_chunks = ((body_len - 1) / CHUNK_SIZE) + 1; + + if (num_chunks > 10000) { + num_chunks = 10000; + } + + NanoHttpBody http_chunks; + http_chunks.bodies_count = num_chunks; + + http_chunks.data = (nano_str_t*)malloc(num_chunks * sizeof(nano_str_t)); + if (!http_chunks.data) { + lua_pushstring(L, "Error: Failed to allocate memory for chunks"); + return lua_error(L); + } + + for (size_t i = 0; i < num_chunks; i++) { + nano_str_t* chunk_ptr = (nano_str_t*)((char*)http_chunks.data + (i * sizeof(nano_str_t))); + size_t chunk_start = i * CHUNK_SIZE; + size_t chunk_len = (i == num_chunks - 1) ? (body_len - chunk_start) : CHUNK_SIZE; + + chunk_ptr->data = (unsigned char*)(body_chunk + chunk_start); + chunk_ptr->len = chunk_len; + } + + AttachmentData attachment_data; + attachment_data.session_id = session_id; + attachment_data.session_data = session_data; + attachment_data.chunk_type = chunk_type; + attachment_data.data = &http_chunks; + + SendDataNanoAttachmentAsync(attachment, &attachment_data); + + free(http_chunks.data); + + lua_pushinteger(L, session_id); + return 1; +} + static int lua_end_inspection(lua_State *L) { NanoAttachment* attachment = (NanoAttachment*) lua_touserdata(L, 1); SessionID session_id = luaL_checkinteger(L, 2); @@ -467,6 +577,46 @@ static int lua_end_inspection(lua_State *L) { return 2; } +static int lua_end_inspection_async(lua_State *L) { + NanoAttachment* attachment = (NanoAttachment*) lua_touserdata(L, 1); + SessionID session_id = luaL_checkinteger(L, 2); + HttpSessionData* session_data = (HttpSessionData*) lua_touserdata(L, 3); + HttpChunkType chunk_type = luaL_checkinteger(L, 4); + + if (!attachment || !session_data) { + lua_pushstring(L, "Error: Invalid attachment or session_data"); + return lua_error(L); + } + + AttachmentData attachment_data; + attachment_data.session_id = session_id; + attachment_data.session_data = session_data; + attachment_data.chunk_type = chunk_type; + attachment_data.data = NULL; + + SendDataNanoAttachmentAsync(attachment, &attachment_data); + + lua_pushinteger(L, session_id); + return 1; +} + +static int lua_get_attachment_verdict_response(lua_State *L) { + NanoAttachment* attachment = (NanoAttachment*) lua_touserdata(L, 1); + SessionID session_id = luaL_checkinteger(L, 2); + + if (!attachment) { + lua_pushstring(L, "Error: Invalid attachment"); + return lua_error(L); + } + + AttachmentVerdictResponse* res_ptr = malloc(sizeof(AttachmentVerdictResponse)); + *res_ptr = getAttachmentVerdictResponse(attachment, session_id); + + lua_pushinteger(L, res_ptr->verdict); + lua_pushlightuserdata(L, res_ptr); + return 2; +} + static int lua_send_response_headers(lua_State *L) { NanoAttachment* attachment = (NanoAttachment*) lua_touserdata(L, 1); SessionID session_id = luaL_checkinteger(L, 2); @@ -507,27 +657,35 @@ static int lua_free_verdict_response(lua_State *L) { return 0; } -static int lua_get_request_processing_timeout_msec(lua_State *L) { +static int lua_get_attachment_socket(lua_State *L) { NanoAttachment* attachment = (NanoAttachment*)lua_touserdata(L, 1); if (!attachment) { - lua_pushinteger(L, 3000); - return 1; + return luaL_error(L, "invalid attachment"); } - uint32_t timeout = GetRequestProcessingTimeout(attachment); - lua_pushinteger(L, timeout); + lua_pushinteger(L, GetCommSocket(attachment)); return 1; } -static int lua_get_response_processing_timeout_msec(lua_State *L) { +static int lua_pop_from_queue(lua_State *L) { NanoAttachment* attachment = (NanoAttachment*)lua_touserdata(L, 1); if (!attachment) { - lua_pushinteger(L, 3000); - return 1; + return luaL_error(L, "invalid attachment"); } - uint32_t timeout = GetResponseProcessingTimeout(attachment); - lua_pushinteger(L, timeout); + SessionID session_id = PopFromNanoQueue(attachment); + lua_pushinteger(L, session_id); + return 1; +} + +static int lua_is_queue_empty(lua_State *L) { + NanoAttachment* attachment = (NanoAttachment*)lua_touserdata(L, 1); + if (!attachment) { + return luaL_error(L, "invalid attachment"); + } + + bool is_empty = isNanoQueueEmpty(attachment); + lua_pushboolean(L, is_empty); return 1; } @@ -552,13 +710,18 @@ static const struct luaL_Reg nano_attachment_lib[] = { {"free_http_metadata", lua_free_http_metadata}, {"free_verdict_response", lua_free_verdict_response}, {"send_body", lua_send_body}, + {"send_body_async", lua_send_body_async}, {"end_inspection", lua_end_inspection}, - {"get_request_processing_timeout_msec", lua_get_request_processing_timeout_msec}, - {"get_response_processing_timeout_msec", lua_get_response_processing_timeout_msec}, + {"end_inspection_async", lua_end_inspection_async}, + {"send_data_async", lua_send_data_async}, + {"get_attachment_verdict_response", lua_get_attachment_verdict_response}, + {"get_attachment_socket", lua_get_attachment_socket}, + {"pop_from_queue", lua_pop_from_queue}, + {"is_queue_empty", lua_is_queue_empty}, {NULL, NULL} }; int luaopen_lua_attachment_wrapper(lua_State *L) { luaL_newlib(L, nano_attachment_lib); return 1; -} \ No newline at end of file +} diff --git a/attachments/kong/plugins/open-appsec-waf-kong-plugin/nano_ffi.lua b/attachments/kong/plugins/open-appsec-waf-kong-plugin/nano_ffi.lua index b88c73d..7311ace 100755 --- a/attachments/kong/plugins/open-appsec-waf-kong-plugin/nano_ffi.lua +++ b/attachments/kong/plugins/open-appsec-waf-kong-plugin/nano_ffi.lua @@ -5,6 +5,7 @@ local nano = {} nano.session_counter = 0 nano.attachments = {} +nano.num_workers = ngx.worker.count() or 1 nano.allocated_strings = {} nano.allocate_headers = {} nano.allocated_metadata = {} @@ -81,14 +82,13 @@ function nano.generate_session_id() return tonumber(string.format("%d%05d", worker_id, nano.session_counter)) end --- Returns: status_code, body, headers -function nano.get_custom_response_data(session_data, response) +function nano.handle_custom_response(session_data, response) local worker_id = ngx.worker.id() local attachment = nano.attachments[worker_id] if not attachment then kong.log.warn("Cannot handle custom response: Attachment not available for worker ", worker_id, " - failing open") - return 200, "Request allowed due to attachment unavailability", {} + return kong.response.exit(200, "Request allowed due to attachment unavailability") end local response_type = nano_attachment.get_web_response_type(attachment, session_data, response) @@ -100,19 +100,18 @@ function nano.get_custom_response_data(session_data, response) code = 403 end kong.log.debug("Response code only: ", code) - return code, "", {} + return kong.response.exit(code, "") end if response_type == nano.WebResponseType.REDIRECT_WEB_RESPONSE then local location = nano_attachment.get_redirect_page(attachment, session_data, response) - local code = nano_attachment.get_response_code(response) or 307 - return code, "", { ["Location"] = location } + return kong.response.exit(307, "", { ["Location"] = location }) end local block_page = nano_attachment.get_block_page(attachment, session_data, response) if not block_page then - kong.log.debug("Failed to retrieve custom block page for session ", session_data) - return 403, "", {} + kong.log.err("Failed to retrieve custom block page for session ", session_data) + return kong.response.exit(500, { message = "Internal Server Error" }) end local code = nano_attachment.get_response_code(response) if not code or code < 100 or code > 599 then @@ -120,13 +119,8 @@ function nano.get_custom_response_data(session_data, response) code = 403 end kong.log.debug("Block page response with code: ", code) - return code, block_page, { ["Content-Type"] = "text/html" } -end + return kong.response.exit(code, block_page, { ["Content-Type"] = "text/html" }) --- Wrapper for backward compatibility - calls kong.response.exit() in access phase -function nano.handle_custom_response(session_data, response) - local code, body, headers = nano.get_custom_response_data(session_data, response) - return kong.response.exit(code, body, headers) end @@ -184,15 +178,15 @@ end function nano.init_attachment() local worker_id = ngx.worker.id() - local num_workers = ngx.worker.count() or 1 -- Get count dynamically local attachment, err local retries = 3 for attempt = 1, retries do - attachment, err = nano_attachment.init_nano_attachment(worker_id, num_workers) + attachment, err = nano_attachment.init_nano_attachment(worker_id, nano.num_workers) if attachment then break end + kong.log.err("Worker ", worker_id, " failed to initialize attachment (attempt ", attempt, "/", retries, "): ", err) end @@ -234,8 +228,8 @@ function nano.is_session_finalized(session_data) local attachment = nano.attachments[worker_id] if not attachment or not session_data then - kong.log.debug("Cannot check session finalization: Invalid attachment or session_data") - return true + kong.log.err("Cannot check session finalization: Invalid attachment or session_data") + return false end return nano_attachment.is_session_finalized(attachment, session_data) @@ -262,6 +256,7 @@ function nano.handle_start_transaction() ) table.insert(nano.allocated_metadata, metadata) + collectgarbage("stop") return metadata @@ -300,6 +295,7 @@ function nano.handleHeaders(headers) end nano_attachment.setHeaderCount(header_data, index) + return header_data end @@ -324,6 +320,21 @@ function nano.send_data(session_id, session_data, meta_data, header_data, contai return verdict, response end +function nano.send_data_async(session_id, session_data, meta_data, header_data, contains_body, chunk_type) + local worker_id = ngx.worker.id() + local attachment = nano.attachments[worker_id] + + if not attachment then + kong.log.warn("Attachment not available for worker ", worker_id, " - failing open") + return nil + end + + contains_body = tonumber(contains_body) or 0 + contains_body = (contains_body > 0) and 1 or 0 + + return nano_attachment.send_data_async(attachment, session_id, session_data, chunk_type, meta_data, header_data, contains_body) +end + function nano.send_body(session_id, session_data, body_chunk, chunk_type) local worker_id = ngx.worker.id() local attachment = nano.attachments[worker_id] @@ -342,6 +353,18 @@ function nano.send_body(session_id, session_data, body_chunk, chunk_type) return verdict, response, modifications end +function nano.send_body_async(session_id, session_data, body_chunk, chunk_type) + local worker_id = ngx.worker.id() + local attachment = nano.attachments[worker_id] + + if not attachment then + kong.log.warn("Attachment not available for worker ", worker_id, " - failing open") + return nil + end + + return nano_attachment.send_body_async(attachment, session_id, session_data, body_chunk, chunk_type) +end + function nano.inject_at_position(buffer, injection, pos) if pos < 0 or pos > #buffer then kong.log.err("Invalid injection position: ", pos, ", buffer length: ", #buffer) @@ -472,7 +495,7 @@ function nano.end_inspection(session_id, session_data, chunk_type) end if not session_data then - kong.log.debug("Cannot end inspection: Invalid session_data for session ", session_id) + kong.log.err("Cannot end inspection: Invalid session_data for session ", session_id) return nano.AttachmentVerdict.INSPECT, nil end @@ -485,30 +508,75 @@ function nano.end_inspection(session_id, session_data, chunk_type) return verdict, response end -function nano.get_request_processing_timeout_sec() +function nano.end_inspection_async(session_id, session_data, chunk_type) local worker_id = ngx.worker.id() local attachment = nano.attachments[worker_id] if not attachment then - kong.log.warn("Attachment not available for worker ", worker_id, " - using default timeout") - return 3 + kong.log.warn("Attachment not available for worker ", worker_id, " - failing open during end_inspection_async") + return nil end - local timeout_msec = nano_attachment.get_request_processing_timeout_msec(attachment) - return timeout_msec / 1000.0 + if not session_data then + kong.log.err("Cannot end inspection async: Invalid session_data for session ", session_id) + return nil + end + + return nano_attachment.end_inspection_async(attachment, session_id, session_data, chunk_type) end -function nano.get_response_processing_timeout_sec() +function nano.get_attachment_verdict_response(session_id) local worker_id = ngx.worker.id() local attachment = nano.attachments[worker_id] if not attachment then - kong.log.warn("Attachment not available for worker ", worker_id, " - using default timeout") - return 3 + kong.log.err("Attachment not available for worker ", worker_id) + return nil, nil end - local timeout_msec = nano_attachment.get_response_processing_timeout_msec(attachment) - return timeout_msec / 1000.0 + local verdict, response = nano_attachment.get_attachment_verdict_response(attachment, session_id) + + if response then + table.insert(nano.allocated_responses, response) + end + + return verdict, response end -return nano \ No newline at end of file +function nano.get_attachment_socket() + local worker_id = ngx.worker.id() + local attachment = nano.attachments[worker_id] + + if not attachment then + kong.log.err("Attachment not available for worker ", worker_id) + return nil + end + + return nano_attachment.get_attachment_socket(attachment) +end + +function nano.pop_from_queue() + local worker_id = ngx.worker.id() + local attachment = nano.attachments[worker_id] + + if not attachment then + kong.log.err("Attachment not available for worker ", worker_id) + return nil + end + + return nano_attachment.pop_from_queue(attachment) +end + +function nano.is_queue_empty() + local worker_id = ngx.worker.id() + local attachment = nano.attachments[worker_id] + + if not attachment then + kong.log.err("Attachment not available for worker ", worker_id) + return true + end + + return nano_attachment.is_queue_empty(attachment) +end + +return nano diff --git a/attachments/kong/plugins/open-appsec-waf-kong-plugin/open-appsec-waf-kong-plugin-1.0.0-1.rockspec b/attachments/kong/plugins/open-appsec-waf-kong-plugin/open-appsec-waf-kong-plugin-1.0.0-1.rockspec index 3b44c02..b7f72da 100755 --- a/attachments/kong/plugins/open-appsec-waf-kong-plugin/open-appsec-waf-kong-plugin-1.0.0-1.rockspec +++ b/attachments/kong/plugins/open-appsec-waf-kong-plugin/open-appsec-waf-kong-plugin-1.0.0-1.rockspec @@ -3,7 +3,7 @@ version = "1.0.0-1" source = { url = "git://github.com/openappsec/attachment.git", - tag = "main" + tag = "feature/async_kong_nano_attachment" } description = { @@ -39,6 +39,8 @@ build = { "attachments/nano_attachment/nano_configuration.c", "attachments/nano_attachment/nano_initializer.c", "attachments/nano_attachment/nano_utils.c", + "attachments/nano_attachment/nano_attachment_sender_async.c", +"attachments/nano_attachment/nano_attachment_bucket.c", "attachments/nano_attachment/nano_attachment_util/nano_attachment_util.cc", "core/attachments/http_configuration/http_configuration.cc", "core/compression/compression_utils.cc", @@ -51,7 +53,13 @@ build = { "external/" }, defines = { "_GNU_SOURCE", "ZLIB_CONST" }, - libraries = { "pthread", "z", "rt", "stdc++" }, + libraries = { "pthread", + "z", + "rt", + "stdc++", + "brotlidec", + "brotlienc", + "brotlicommon"}, ldflags = { "-static-libstdc++", "-static-libgcc" } } } diff --git a/attachments/kong/plugins/open-appsec-waf-kong-plugin/schema.lua b/attachments/kong/plugins/open-appsec-waf-kong-plugin/schema.lua index 0d722ee..b044c0c 100755 --- a/attachments/kong/plugins/open-appsec-waf-kong-plugin/schema.lua +++ b/attachments/kong/plugins/open-appsec-waf-kong-plugin/schema.lua @@ -3,19 +3,12 @@ local typedefs = require "kong.db.schema.typedefs" return { name = "open-appsec-waf-kong-plugin", fields = { - { consumer = typedefs.no_consumer }, -- required for Konnect compatibility - { - protocols = { - type = "set", - elements = { type = "string", one_of = { "http", "https" } }, - default = { "http", "https" }, - }, - }, - { - config = { + { consumer = typedefs.no_consumer }, -- required for Konnect compatibility + { protocols = typedefs.protocols_http }, -- required so Konnect knows when to allow this plugin + { config = { type = "record", fields = { - { debug = { type = "boolean", default = false } }, + { debug = { type = "boolean", default = false } }, }, }, },