Modified Kong Plugin to support the new Nano Attachment

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()
+NanoHandler.sessions = {}
-    nano.init_attachment()
+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
 
-function NanoHandler.access(conf)
+local function start_verdict_listener()
-    local ctx = kong.ctx.plugin
+    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)
+    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
+        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)
 
-    ctx.session_data = session_data
+    kong.ctx.plugin.session_data = session_data
-    ctx.session_id = session_id
+    kong.ctx.plugin.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")
-        return
-    end
 
     local meta_data = nano.handle_start_transaction()
     if not meta_data then
-        kong.log.debug("Failed to handle start transaction - failing mode")
+        kong.log.err("access: Failed to handle start transaction for session_id=", session_id, " - failing open")
-        ctx.cleanup_needed = true
         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)
+    local sem = semaphore.new()
-    if verdict ~= nano.AttachmentVerdict.INSPECT then
+    pending[session_id] = { sem = sem, verdict = nil }
-        ctx.cleanup_needed = true
+
-        if verdict == nano.AttachmentVerdict.DROP then
+    -- Use non-blocking send_data_async
-            return nano.handle_custom_response(session_data, response)
+    kong.log.debug("access: Sending headers async for session_id=", session_id, " contains_body=", contains_body)
-        end
+    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)
+            -- Use non-blocking send_body_async
-            if verdict ~= nano.AttachmentVerdict.INSPECT then
+            kong.log.debug("access: Sending body async, size=", #body, " for session_id=", session_id)
-                ctx.cleanup_needed = true
+            nano.send_body_async(session_id, session_data, body, nano.HttpChunkType.HTTP_REQUEST_BODY)
-                if verdict == nano.AttachmentVerdict.DROP then
+
-                    return nano.handle_custom_response(session_data, response)
+            -- Wait on semaphore for verdict
-                end
+            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
+
+            -- 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
                 local body_file = ngx.var.request_body_file
                 if body_file then
-                local file = io.open(body_file, "rb")
+                    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 chunk_size = 8192
+                        local entire_body = file:read("*all")
-                    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 entire_body and #entire_body > 0 then
-                        if not chunk or #chunk == 0 then
+                            kong.log.debug("access: Sending entire body from file, size=", #entire_body, " bytes for session_id=", session_id)
-                            kong.log.debug("End of request body file reached")
+                            verdict, response = nano.send_body(session_id, session_data, entire_body, nano.HttpChunkType.HTTP_REQUEST_BODY)
-                            break
+                            kong.log.debug("access: File body verdict=", verdict, " for session_id=", session_id)
-                        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)
+                                kong.log.warn("access: File body verdict DROP for session_id=", session_id)
-                            end
+                                nano.fini_session(session_data)
-                            return
+                                kong.ctx.plugin.blocked = true
-                        end
+                                local result = nano.handle_custom_response(session_data, response)
-                    end
+                                nano.cleanup_all()
-                    file:close()
+                                return result
-                    kong.log.debug("Sent ", chunk_count, " chunks from request body file")
                             end
                         else
-                kong.log.err("Request body expected but no body data or file available")
+                            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()
+        kong.log.debug("access: Ending request inspection (with body) for session_id=", session_id)
-            return nano.end_inspection(session_id, session_data, nano.HttpChunkType.HTTP_REQUEST_END)
+        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")
+            kong.log.err("access: Error ending request inspection for session_id=", session_id, " err=", result, " - failing open")
-            ctx.cleanup_needed = true
+            nano.fini_session(session_data)
-            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
-            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
-
-    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)
-        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)
-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
-
-    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
-
-    if chunk and #chunk > 0 then
-        ctx.body_buffer_chunk = ctx.body_buffer_chunk or 0
-        ctx.body_seen = true
-
-        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
-
-        ctx.body_buffer_chunk = ctx.body_buffer_chunk + 1
-
-        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
-
-    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.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
+            return
-        ctx.session_id = nil
-        collectgarbage("collect")
         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
+
+        -- 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 verdict == nano.AttachmentVerdict.DROP then
+            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
+    end
+
+    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.header_filter(conf)
+--     local ctx = kong.ctx.plugin
+--     if ctx.blocked then
+--         return
+--     end
+
+--     local session_id = ctx.session_id
+--     local session_data = ctx.session_data
+
+--     if not session_id or not session_data then
+--         return
+--     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
+
+--     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
+
+--     ctx.expect_body = not (status_code == 204 or status_code == 304 or (100 <= status_code and status_code < 200) or content_length == 0)
+-- end
+
+-- function NanoHandler.body_filter(conf)
+--     local ctx = kong.ctx.plugin
+--     if ctx.blocked then
+--         return
+--     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
+

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 luaL_error(L, "invalid attachment");
-        return 1;
     }
 
-    uint32_t timeout = GetRequestProcessingTimeout(attachment);
+    lua_pushinteger(L, GetCommSocket(attachment));
-    lua_pushinteger(L, timeout);
     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 luaL_error(L, "invalid attachment");
-        return 1;
     }
 
-    uint32_t timeout = GetResponseProcessingTimeout(attachment);
+    SessionID session_id = PopFromNanoQueue(attachment);
-    lua_pushinteger(L, timeout);
+    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,9 +710,14 @@ 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},
+    {"end_inspection_async", lua_end_inspection_async},
-    {"get_response_processing_timeout_msec", lua_get_response_processing_timeout_msec},
+    {"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}
 };
 

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.handle_custom_response(session_data, response)
-function nano.get_custom_response_data(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 kong.response.exit(307, "", { ["Location"] = location })
-        return code, "", { ["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)
+        kong.log.err("Failed to retrieve custom block page for session ", session_data)
-        return 403, "", {}
+        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" }
+    return kong.response.exit(code, block_page, { ["Content-Type"] = "text/html" })
-end
 
--- 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")
+        kong.log.err("Cannot check session finalization: Invalid attachment or session_data")
-        return true
+        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")
+        kong.log.warn("Attachment not available for worker ", worker_id, " - failing open during end_inspection_async")
-        return 3
+        return nil
     end
 
-    local timeout_msec = nano_attachment.get_request_processing_timeout_msec(attachment)
+    if not session_data then
-    return timeout_msec / 1000.0
+        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")
+        kong.log.err("Attachment not available for worker ", worker_id)
-        return 3
+        return nil, nil
     end
 
-    local timeout_msec = nano_attachment.get_response_processing_timeout_msec(attachment)
+    local verdict, response = nano_attachment.get_attachment_verdict_response(attachment, session_id)
-    return timeout_msec / 1000.0
+
+    if response then
+        table.insert(nano.allocated_responses, response)
+    end
+
+    return verdict, response
+end
+
+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

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" }
     }
   }

attachments/kong/plugins/open-appsec-waf-kong-plugin/schema.lua

@@ -4,15 +4,8 @@ return {
   name = "open-appsec-waf-kong-plugin",
   fields = {
     { consumer = typedefs.no_consumer },                -- required for Konnect compatibility
-    {
+    { protocols = typedefs.protocols_http },            -- required so Konnect knows when to allow this plugin
-      protocols = {
+    { config = {
-        type = "set",
-        elements = { type = "string", one_of = { "http", "https" } },
-        default = { "http", "https" },
-      },
-    },
-    {
-      config = {
         type = "record",
         fields = {
           { debug  = { type = "boolean", default = false } },