mirror of
https://github.com/Ed1s0nZ/CyberStrikeAI.git
synced 2026-03-31 00:09:29 +02:00
615 lines
22 KiB
YAML
615 lines
22 KiB
YAML
name: "http-framework-test"
|
||
command: "python3"
|
||
args:
|
||
- "-c"
|
||
- |
|
||
import argparse
|
||
import json
|
||
import re
|
||
import shlex
|
||
import subprocess
|
||
import sys
|
||
import time
|
||
import urllib.parse
|
||
|
||
try:
|
||
from charset_normalizer import from_bytes as charset_from_bytes
|
||
except ImportError:
|
||
charset_from_bytes = None
|
||
|
||
try:
|
||
import chardet
|
||
except ImportError:
|
||
chardet = None
|
||
|
||
METRIC_MARKER = "__CYBERSTRIKE_HTTP_METRICS__"
|
||
METRIC_KEYS = [
|
||
"dns_lookup",
|
||
"tcp_connect",
|
||
"tls_handshake",
|
||
"pretransfer",
|
||
"ttfb",
|
||
"total",
|
||
"speed_download",
|
||
"size_download",
|
||
"http_code",
|
||
"redirects",
|
||
]
|
||
|
||
def parse_headers(raw: str):
|
||
if not raw:
|
||
return []
|
||
raw = raw.strip()
|
||
if not raw:
|
||
return []
|
||
try:
|
||
parsed = json.loads(raw)
|
||
headers = []
|
||
if isinstance(parsed, dict):
|
||
for k, v in parsed.items():
|
||
headers.append(f"{k}: {v}")
|
||
return headers
|
||
if isinstance(parsed, list):
|
||
for item in parsed:
|
||
if isinstance(item, str) and item.strip():
|
||
headers.append(item.strip())
|
||
if headers:
|
||
return headers
|
||
except json.JSONDecodeError:
|
||
pass
|
||
headers = []
|
||
for line in raw.replace(";", "\n").splitlines():
|
||
stripped = line.strip()
|
||
if stripped:
|
||
headers.append(stripped)
|
||
return headers
|
||
|
||
def parse_additional(raw: str):
|
||
if not raw:
|
||
return []
|
||
try:
|
||
return shlex.split(raw)
|
||
except ValueError:
|
||
return [arg for arg in raw.split() if arg]
|
||
|
||
def smart_encode_url(url: str, safe_path="/:@&=%+,$-~", safe_query="/:@&=%+,$-~"):
|
||
try:
|
||
parts = urllib.parse.urlsplit(url)
|
||
except ValueError:
|
||
return url
|
||
path = urllib.parse.quote(parts.path or "/", safe=safe_path)
|
||
query = urllib.parse.quote(parts.query, safe=safe_query)
|
||
fragment = urllib.parse.quote(parts.fragment, safe=safe_query)
|
||
return urllib.parse.urlunsplit((parts.scheme, parts.netloc, path, query, fragment))
|
||
|
||
def sanitize_cmd(cmd):
|
||
return " ".join(shlex.quote(part) for part in cmd)
|
||
|
||
def extract_metrics(output: bytes):
|
||
marker = (METRIC_MARKER + ":").encode("ascii")
|
||
if marker not in output:
|
||
return output, {}
|
||
head, tail = output.rsplit(marker, 1)
|
||
try:
|
||
tail_text = tail.decode("utf-8", errors="ignore")
|
||
except UnicodeDecodeError:
|
||
tail_text = ""
|
||
values = tail_text.strip().split("|")
|
||
stats = {}
|
||
for key, value in zip(METRIC_KEYS, values):
|
||
stats[key] = value.strip()
|
||
return head, stats
|
||
|
||
def extract_declared_charset(data: bytes):
|
||
if not data:
|
||
return ""
|
||
sample = data[:16384]
|
||
try:
|
||
sample_text = sample.decode("iso-8859-1", errors="ignore")
|
||
except UnicodeDecodeError:
|
||
return ""
|
||
|
||
for line in sample_text.splitlines():
|
||
lowered = line.lower()
|
||
if "content-type:" in lowered and "charset=" in lowered:
|
||
charset_index = lowered.index("charset=") + len("charset=")
|
||
remainder = line[charset_index:]
|
||
for separator in [";", " ", "\t"]:
|
||
remainder = remainder.split(separator)[0]
|
||
return remainder.strip().strip('"').strip("'")
|
||
|
||
meta_match = re.search(r'charset=["\']?([a-zA-Z0-9_\-.:]+)', sample_text, re.IGNORECASE)
|
||
if meta_match:
|
||
return meta_match.group(1)
|
||
return ""
|
||
|
||
def decode_body_bytes(data: bytes, user_encoding: str = ""):
|
||
declared = extract_declared_charset(data)
|
||
attempts = []
|
||
if user_encoding:
|
||
attempts.append(("user", user_encoding))
|
||
if declared:
|
||
attempts.append(("declared", declared))
|
||
|
||
for source, encoding in attempts:
|
||
enc = (encoding or "").strip()
|
||
if not enc:
|
||
continue
|
||
try:
|
||
return data.decode(enc), enc, source
|
||
except (LookupError, UnicodeDecodeError):
|
||
continue
|
||
|
||
if charset_from_bytes is not None and data:
|
||
best = charset_from_bytes(data).best()
|
||
if best and best.encoding:
|
||
try:
|
||
return data.decode(best.encoding), best.encoding, "detected"
|
||
except (LookupError, UnicodeDecodeError):
|
||
pass
|
||
|
||
if chardet is not None and data:
|
||
detection = chardet.detect(data)
|
||
encoding = detection.get("encoding")
|
||
if encoding:
|
||
try:
|
||
return data.decode(encoding), encoding, "detected"
|
||
except (LookupError, UnicodeDecodeError):
|
||
pass
|
||
|
||
try:
|
||
return data.decode("utf-8"), "utf-8", "fallback"
|
||
except UnicodeDecodeError:
|
||
return data.decode("utf-8", errors="replace"), "utf-8", "fallback"
|
||
|
||
def to_float(value):
|
||
try:
|
||
return float(value)
|
||
except (TypeError, ValueError):
|
||
return None
|
||
|
||
def encode_form_data(data: str):
|
||
"""
|
||
对application/x-www-form-urlencoded格式的数据进行URL编码
|
||
解析key=value格式，对值部分进行URL编码，保持键名不变
|
||
支持多个键值对（用&分隔）
|
||
正确处理值中包含=的情况（使用split('=', 1)只分割第一个=）
|
||
智能处理值中包含&的情况（通过检查&后是否跟着key=模式来判断）
|
||
|
||
注意：假设输入是未编码的原始数据，如果已经是编码的，可能会重复编码
|
||
"""
|
||
if not data:
|
||
return data
|
||
|
||
# 智能分割：找到真正的键值对分隔符&
|
||
# &是分隔符的条件：&后面跟着key=模式（即非空白字符后跟=）
|
||
def find_key_value_pairs(text):
|
||
"""找到所有的key=value对，正确处理值中的&和="""
|
||
pairs = []
|
||
i = 0
|
||
text_len = len(text)
|
||
|
||
while i < text_len:
|
||
# 跳过空白
|
||
while i < text_len and text[i] in ' \t\n\r':
|
||
i += 1
|
||
if i >= text_len:
|
||
break
|
||
|
||
# 查找键名（到第一个=为止）
|
||
key_start = i
|
||
while i < text_len and text[i] != '=':
|
||
i += 1
|
||
|
||
if i >= text_len:
|
||
# 没有=，可能是单个值
|
||
remaining = text[key_start:].strip()
|
||
if remaining:
|
||
pairs.append((None, remaining))
|
||
break
|
||
|
||
# 提取键名
|
||
key = text[key_start:i].strip()
|
||
i += 1 # 跳过=
|
||
|
||
if not key:
|
||
continue
|
||
|
||
# 查找值的结束位置
|
||
# 值可能包含&，需要判断&是否是分隔符
|
||
value_start = i
|
||
value_end = text_len
|
||
|
||
# 从值开始位置查找&
|
||
j = value_start
|
||
while j < text_len:
|
||
if text[j] == '&':
|
||
# 检查&后面是否跟着key=模式
|
||
k = j + 1
|
||
# 跳过空白
|
||
while k < text_len and text[k] in ' \t\n\r':
|
||
k += 1
|
||
# 查找下一个=或&
|
||
m = k
|
||
while m < text_len and text[m] not in '=&':
|
||
m += 1
|
||
# 如果找到=，说明&后面是新的键值对
|
||
if m < text_len and text[m] == '=':
|
||
value_end = j
|
||
i = j + 1 # 从&后开始下一轮
|
||
break
|
||
j += 1
|
||
|
||
# 提取值
|
||
value = text[value_start:value_end]
|
||
pairs.append((key, value))
|
||
|
||
if value_end < text_len:
|
||
i = value_end + 1
|
||
else:
|
||
break
|
||
|
||
return pairs
|
||
|
||
# 解析所有键值对
|
||
pairs = find_key_value_pairs(data)
|
||
parts = []
|
||
|
||
for key, value in pairs:
|
||
if key is None:
|
||
# 没有键，只有值
|
||
parts.append(urllib.parse.quote_plus(value, safe=''))
|
||
else:
|
||
# 对值进行URL编码
|
||
encoded_value = urllib.parse.quote_plus(value, safe='')
|
||
parts.append(f"{key}={encoded_value}")
|
||
|
||
return '&'.join(parts)
|
||
|
||
def should_encode_data(headers: list, data: str):
|
||
"""
|
||
判断是否需要对POST数据进行URL编码
|
||
如果Content-Type是application/x-www-form-urlencoded，则需要编码
|
||
"""
|
||
if not data:
|
||
return False
|
||
|
||
content_type = None
|
||
for header in headers:
|
||
if ':' in header:
|
||
h_key, h_value = header.split(':', 1)
|
||
if h_key.strip().lower() == 'content-type':
|
||
content_type = h_value.strip().lower()
|
||
break
|
||
|
||
if content_type and 'application/x-www-form-urlencoded' in content_type:
|
||
return True
|
||
return False
|
||
|
||
parser = argparse.ArgumentParser(description="Enhanced HTTP testing helper")
|
||
parser.add_argument("--url", required=True)
|
||
parser.add_argument("--method", default="GET")
|
||
parser.add_argument("--data", default="")
|
||
parser.add_argument("--headers", default="")
|
||
parser.add_argument("--cookies", default="")
|
||
parser.add_argument("--user-agent", dest="user_agent", default="")
|
||
parser.add_argument("--proxy", default="")
|
||
parser.add_argument("--timeout", default="")
|
||
parser.add_argument("--repeat", type=int, default=1)
|
||
parser.add_argument("--delay", default="0")
|
||
parser.add_argument("--additional-args", dest="additional_args", default="")
|
||
parser.add_argument("--action", default="")
|
||
parser.add_argument("--include-headers", dest="include_headers", action="store_true")
|
||
parser.add_argument("--auto-encode-url", dest="auto_encode_url", action="store_true")
|
||
parser.add_argument("--follow-redirects", dest="follow_redirects", action="store_true")
|
||
parser.add_argument("--allow-insecure", dest="allow_insecure", action="store_true")
|
||
parser.add_argument("--verbose-output", dest="verbose_output", action="store_true")
|
||
parser.add_argument("--show-command", dest="show_command", action="store_true")
|
||
parser.add_argument("--show-summary", dest="show_summary", action="store_true")
|
||
parser.add_argument("--response-encoding", dest="response_encoding", default="")
|
||
parser.add_argument("--debug", dest="debug", action="store_true")
|
||
args = parser.parse_args()
|
||
|
||
repeat = max(1, args.repeat)
|
||
try:
|
||
delay_between = float(args.delay or "0")
|
||
if delay_between < 0:
|
||
delay_between = 0.0
|
||
except ValueError:
|
||
delay_between = 0.0
|
||
|
||
prepared_url = smart_encode_url(args.url) if args.auto_encode_url else args.url
|
||
curl_cmd = ["curl", "-sS"]
|
||
if args.include_headers:
|
||
curl_cmd.append("-i")
|
||
if args.verbose_output:
|
||
curl_cmd.append("-v")
|
||
method = (args.method or "GET").upper()
|
||
if method:
|
||
curl_cmd.extend(["-X", method])
|
||
if args.cookies:
|
||
curl_cmd.extend(["-b", args.cookies])
|
||
if args.user_agent:
|
||
curl_cmd.extend(["-A", args.user_agent])
|
||
if args.timeout:
|
||
curl_cmd.extend(["--max-time", str(args.timeout)])
|
||
if args.follow_redirects:
|
||
curl_cmd.append("-L")
|
||
if args.allow_insecure:
|
||
curl_cmd.append("-k")
|
||
if args.proxy:
|
||
curl_cmd.extend(["-x", args.proxy])
|
||
|
||
# 解析headers以便检查Content-Type
|
||
parsed_headers = parse_headers(args.headers)
|
||
for header in parsed_headers:
|
||
curl_cmd.extend(["-H", header])
|
||
|
||
# 处理POST数据：如果是表单数据，需要URL编码
|
||
prepared_data = args.data
|
||
if args.debug and args.data:
|
||
print("\n===== Debug: POST Data Processing =====")
|
||
print(f"Original data: {repr(args.data)}")
|
||
print(f"Data length: {len(args.data)} bytes")
|
||
# 显示原始数据中的键值对
|
||
if '=' in args.data:
|
||
parts = args.data.split('&')
|
||
print(f"Detected {len(parts)} key-value pair(s):")
|
||
for i, part in enumerate(parts, 1):
|
||
if '=' in part:
|
||
k, v = part.split('=', 1)
|
||
print(f" [{i}] {k} = {repr(v[:50])}{'...' if len(v) > 50 else ''} (length: {len(v)})")
|
||
else:
|
||
print(f" [{i}] (no key): {repr(part[:50])}{'...' if len(part) > 50 else ''}")
|
||
if should_encode_data(parsed_headers, args.data):
|
||
print("Content-Type detected: application/x-www-form-urlencoded")
|
||
print("Encoding will be applied...")
|
||
else:
|
||
print("No encoding needed (not form-urlencoded)")
|
||
|
||
if args.data and should_encode_data(parsed_headers, args.data):
|
||
prepared_data = encode_form_data(args.data)
|
||
if args.debug:
|
||
print(f"\nEncoded data: {repr(prepared_data)}")
|
||
print(f"Encoded length: {len(prepared_data)} bytes")
|
||
# 显示编码后的键值对
|
||
if '&' in prepared_data:
|
||
encoded_parts = prepared_data.split('&')
|
||
print(f"Encoded into {len(encoded_parts)} key-value pair(s):")
|
||
for i, part in enumerate(encoded_parts, 1):
|
||
if '=' in part:
|
||
k, v = part.split('=', 1)
|
||
print(f" [{i}] {k} = {repr(v[:80])}{'...' if len(v) > 80 else ''} (length: {len(v)})")
|
||
else:
|
||
print(f" [{i}] (no key): {repr(part[:80])}{'...' if len(part) > 80 else ''}")
|
||
|
||
if prepared_data:
|
||
curl_cmd.extend(["--data", prepared_data])
|
||
metrics_template = METRIC_MARKER + ":" + "|".join([
|
||
"%{time_namelookup}",
|
||
"%{time_connect}",
|
||
"%{time_appconnect}",
|
||
"%{time_pretransfer}",
|
||
"%{time_starttransfer}",
|
||
"%{time_total}",
|
||
"%{speed_download}",
|
||
"%{size_download}",
|
||
"%{http_code}",
|
||
"%{num_redirects}",
|
||
])
|
||
curl_cmd.extend(["-w", f"\n{metrics_template}\n"])
|
||
if args.additional_args:
|
||
curl_cmd.extend(parse_additional(args.additional_args))
|
||
curl_cmd.append(prepared_url)
|
||
|
||
aggregate = {key: [] for key in METRIC_KEYS} if args.show_summary else None
|
||
if aggregate is not None:
|
||
aggregate["wall_time"] = []
|
||
exit_code = 0
|
||
|
||
for run_index in range(repeat):
|
||
if run_index > 0 and delay_between > 0:
|
||
time.sleep(delay_between)
|
||
|
||
run_cmd = list(curl_cmd)
|
||
start = time.perf_counter()
|
||
proc = subprocess.run(
|
||
run_cmd,
|
||
capture_output=True,
|
||
)
|
||
elapsed = time.perf_counter() - start
|
||
body_bytes, stats = extract_metrics(proc.stdout)
|
||
decoded_body, used_encoding, encoding_source = decode_body_bytes(
|
||
body_bytes,
|
||
user_encoding=args.response_encoding,
|
||
)
|
||
|
||
print(f"\n===== Response #{run_index + 1} =====")
|
||
output_body = decoded_body.rstrip()
|
||
if output_body:
|
||
print(output_body)
|
||
else:
|
||
print("[no body]")
|
||
|
||
print(f"\n----- Meta #{run_index + 1} -----")
|
||
if args.show_command and run_index == 0:
|
||
print("Command:", sanitize_cmd(run_cmd))
|
||
if args.show_summary:
|
||
if stats:
|
||
for key in METRIC_KEYS:
|
||
label = key.replace("_", " ").title()
|
||
print(f"{label}: {stats.get(key, 'n/a')}")
|
||
value = to_float(stats.get(key))
|
||
if value is not None and aggregate is not None:
|
||
aggregate[key].append(value)
|
||
else:
|
||
print("Timing data unavailable (curl -w output missing).")
|
||
print(f"Wall Time (client): {elapsed:.6f}s")
|
||
if aggregate is not None:
|
||
aggregate["wall_time"].append(elapsed)
|
||
else:
|
||
print("Summary disabled (--show-summary=false).")
|
||
print(f"Wall Time (client): {elapsed:.6f}s")
|
||
|
||
print(f"Encoding Used: {used_encoding} ({encoding_source})")
|
||
|
||
stderr_text = proc.stderr.decode("utf-8", errors="replace").strip()
|
||
if stderr_text:
|
||
print("\nstderr:")
|
||
print(stderr_text)
|
||
|
||
if proc.returncode != 0:
|
||
exit_code = proc.returncode
|
||
|
||
if args.show_summary and repeat > 1 and aggregate is not None:
|
||
def summarize(values):
|
||
if not values:
|
||
return None
|
||
return (min(values), sum(values)/len(values), max(values))
|
||
|
||
print("\n===== Aggregate Timing =====")
|
||
for key, values in aggregate.items():
|
||
summary = summarize(values)
|
||
if not summary:
|
||
continue
|
||
label = key.replace("_", " ").title()
|
||
min_v, avg_v, max_v = summary
|
||
print(f"{label}: min {min_v:.6f}s | avg {avg_v:.6f}s | max {max_v:.6f}s")
|
||
|
||
if exit_code != 0:
|
||
sys.exit(exit_code)
|
||
sys.exit(0)
|
||
enabled: true
|
||
short_description: "增强的HTTP测试框架（带延时、编码、可观察性）"
|
||
description: |
|
||
增强的HTTP测试框架：提供自动URL编码、详细响应/时延输出、重复请求和命令可见性，可用于常规请求、重放、盲注延时观测等场景。
|
||
|
||
**能力亮点：**
|
||
- 自动URL编码：解决包含空格、引号等字符时curl报错的问题，必要时可手动关闭
|
||
- 时延观测：采集 DNS / TCP / TLS / TTFB / 总耗时，可循环请求计算盲注延时
|
||
- 详细输出：可选响应头、命令、stderr，方便排查
|
||
- 自动编码识别：支持响应内容编码自动检测，可额外指定强制编码
|
||
- 扩展控制：支持代理、超时、重复次数、延迟间隔及原生curl参数透传
|
||
parameters:
|
||
- name: "url"
|
||
type: "string"
|
||
description: "目标URL（自动进行路径/查询编码，确保特殊字符安全发送）"
|
||
required: true
|
||
flag: "--url"
|
||
- name: "method"
|
||
type: "string"
|
||
description: "HTTP方法（GET, POST, PUT, DELETE等）"
|
||
required: false
|
||
default: "GET"
|
||
flag: "--method"
|
||
- name: "data"
|
||
type: "string"
|
||
description: "请求数据/参数（JSON、表单、原始payload均可）"
|
||
required: false
|
||
flag: "--data"
|
||
- name: "headers"
|
||
type: "string"
|
||
description: "自定义请求头（JSON字典、行分隔或以分号分隔的 Header: Value 格式）"
|
||
required: false
|
||
flag: "--headers"
|
||
- name: "cookies"
|
||
type: "string"
|
||
description: "自定义Cookie（格式：name1=value1; name2=value2）"
|
||
required: false
|
||
flag: "--cookies"
|
||
- name: "user_agent"
|
||
type: "string"
|
||
description: "自定义User-Agent"
|
||
required: false
|
||
flag: "--user-agent"
|
||
- name: "proxy"
|
||
type: "string"
|
||
description: "代理（curl -x 形式，如 http://127.0.0.1:8080）"
|
||
required: false
|
||
flag: "--proxy"
|
||
- name: "timeout"
|
||
type: "string"
|
||
description: "最大超时时间（秒，传递给curl --max-time）"
|
||
required: false
|
||
flag: "--timeout"
|
||
- name: "repeat"
|
||
type: "int"
|
||
description: "重复请求次数，用于盲注延时观测（>=1）"
|
||
required: false
|
||
default: 1
|
||
flag: "--repeat"
|
||
- name: "delay"
|
||
type: "string"
|
||
description: "重复请求之间的延迟（秒，可为小数）"
|
||
required: false
|
||
default: "0"
|
||
flag: "--delay"
|
||
- name: "include_headers"
|
||
type: "bool"
|
||
description: "输出响应头（等价于curl -i），默认开启"
|
||
required: false
|
||
default: true
|
||
flag: "--include-headers"
|
||
- name: "auto_encode_url"
|
||
type: "bool"
|
||
description: "自动URL编码（默认开启，避免出现URL格式错误）"
|
||
required: false
|
||
default: true
|
||
flag: "--auto-encode-url"
|
||
- name: "follow_redirects"
|
||
type: "bool"
|
||
description: "跟随重定向（curl -L）"
|
||
required: false
|
||
default: false
|
||
flag: "--follow-redirects"
|
||
- name: "allow_insecure"
|
||
type: "bool"
|
||
description: "忽略TLS证书错误（curl -k）"
|
||
required: false
|
||
default: false
|
||
flag: "--allow-insecure"
|
||
- name: "verbose_output"
|
||
type: "bool"
|
||
description: "输出curl调试信息（curl -v）"
|
||
required: false
|
||
default: false
|
||
flag: "--verbose-output"
|
||
- name: "show_command"
|
||
type: "bool"
|
||
description: "打印最终curl命令（含自动编码后的URL），默认开启"
|
||
required: false
|
||
default: true
|
||
flag: "--show-command"
|
||
- name: "show_summary"
|
||
type: "bool"
|
||
description: "打印高亮摘要（默认开启）"
|
||
required: false
|
||
default: true
|
||
flag: "--show-summary"
|
||
- name: "debug"
|
||
type: "bool"
|
||
description: "调试模式：打印POST数据的原始值和编码后的值，方便排查编码问题"
|
||
required: false
|
||
default: false
|
||
flag: "--debug"
|
||
- name: "response_encoding"
|
||
type: "string"
|
||
description: "强制响应解码使用的编码（默认自动检测，可用于指定GBK等场景）"
|
||
required: false
|
||
flag: "--response-encoding"
|
||
- name: "action"
|
||
type: "string"
|
||
description: "保留字段：标识调用意图（request, spider等），当前脚本内部不使用"
|
||
required: false
|
||
default: "request"
|
||
- name: "additional_args"
|
||
type: "string"
|
||
description: |
|
||
额外的curl参数（原样透传），多个参数使用空格或带引号的shell风格传入。
|
||
|
||
**示例值：**
|
||
- "-H 'Origin: https://target'"
|
||
- "--interface tun0 --compressed"
|
||
required: false
|
||
flag: "--additional-args"
|