Python for Pentesters: Networking, HTTP, Scapy, and Security Automation

Pentesters use Python to glue tools together: read a target list, call a scanner, parse the result, hit an HTTP API, and write a clean JSON report. This lesson covers those primitives and one small reusable checker.

What Python skills a pentester actually needs

You do not need a full CS syllabus to be useful on an engagement. Spend time on:

• Files and paths (pathlib, open)
• Structured data (json, CSV)
• Regex for log and banner scraps
• subprocess to drive CLI tools safely
• Sockets and HTTP clients
• CLI arguments (argparse)
• Timeouts, logging, and exit codes

Types, loops, and functions matter, but only as the glue for those jobs. If you can write a 40-line script that reads targets, runs a check with a timeout, and writes JSON, you are already ahead of most “port scanner from scratch” tutorials.

Set up the Python pentest lab on Kali

Use a disposable working directory so scrap files stay out of your home clutter:

mkdir -p ~/python-pentest-lab/{scripts,data,out} && cd ~/python-pentest-lab

Confirm the interpreter and libraries this article used:

python3 --version

Python 3.13.12

Confirm requests and Scapy import from this interpreter:

python3 -c 'import requests, scapy; print(requests.__version__, scapy.__version__)'

2.32.5 2.7.01

scapy.__version__ prints 2.7.01 on this Kali package even though dpkg lists python3-scapy 2.7.0+dfsg1-1. Trust the import you actually run.

Start a local HTTP teaching app

Save http_lab.py (loopback only):

#!/usr/bin/env python3
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
import json

class H(BaseHTTPRequestHandler):
    def _json(self, code, obj):
        body = json.dumps(obj).encode()
        self.send_response(code)
        self.send_header('Content-Type', 'application/json')
        self.send_header('X-Lab-App', 'python-pentest-lab')
        self.send_header('Set-Cookie', 'lab_session=demo; Path=/')
        self.send_header('Content-Length', str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def do_GET(self):
        if self.path.startswith('/api/health'):
            self._json(200, {'status': 'ok', 'path': self.path})
            return
        if self.path.startswith('/api/headers'):
            self._json(200, {'headers': dict(self.headers)})
            return
        self.send_error(404)

    def do_POST(self):
        length = int(self.headers.get('Content-Length', 0))
        raw = self.rfile.read(length) if length else b''
        try:
            data = json.loads(raw.decode() or '{}')
        except Exception:
            data = {'_raw': raw.decode(errors='replace')}
        if self.path.startswith('/api/echo'):
            self._json(200, {'echo': data, 'cookie': self.headers.get('Cookie')})
            return
        self.send_error(404)

    def log_message(self, fmt, *args):
        return

if __name__ == '__main__':
    ThreadingHTTPServer(('127.0.0.1', 42020), H).serve_forever()

Start it in another terminal (python3 http_lab.py). It stays quiet until you send requests.

Start a local TCP echo service

Save tcp_echo.py:

#!/usr/bin/env python3
import socket

HOST, PORT = '127.0.0.1', 42021
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
    s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    s.bind((HOST, PORT))
    s.listen(5)
    print(f'listening on {HOST}:{PORT}', flush=True)
    while True:
        conn, addr = s.accept()
        with conn:
            data = conn.recv(4096)
            if data:
                conn.sendall(b'ECHO:' + data)

Start it with python3 tcp_echo.py. You should see listening on 127.0.0.1:42021. Keep both listeners running for the rest of the lesson.

Automate safe input and output

Pentest scripts should read structured input and write structured output. Avoid printing only human text you cannot parse later.

Write a tiny target list:

printf '%s\n' '127.0.0.1' > data/hosts.txt

Read it and emit JSON lines:

python3 - <<'PY'
from pathlib import Path
import json
hosts = [line.strip() for line in Path('data/hosts.txt').read_text().splitlines() if line.strip()]
Path('out/hosts.json').write_text(json.dumps({'hosts': hosts}, indent=2) + '\n')
print(Path('out/hosts.json').read_text())
PY

{
  "hosts": [
    "127.0.0.1"
  ]
}

That pattern scales: load hosts from a file the RoE already approved, then attach results per host.

Drive a CLI tool with subprocess instead of os.system. Capture stdout, set a timeout, and fail loudly:

python3 - <<'PY'
import subprocess
out = subprocess.run(
    ['nmap', '-p', '42020,42021', '127.0.0.1'],
    capture_output=True,
    text=True,
    timeout=30,
    check=True,
)
print(out.stdout)
PY

Starting Nmap 7.99 ( https://nmap.org ) at 2026-09-05 05:57 -0400
Nmap scan report for localhost (127.0.0.1)
Host is up (0.000066s latency).

PORT      STATE SERVICE
42020/tcp open  unknown
42021/tcp open  unknown

Nmap done: 1 IP address (1 host up) scanned in 0.09 seconds

Both lab ports show open. Prefer subprocess.run([...], check=True) over shell strings so user-controlled input cannot inject extra commands.

Use Python requests for HTTP checks

With http_lab.py running, probe health, headers, cookies, and JSON POST bodies.

python3 - <<'PY'
import requests
BASE = 'http://127.0.0.1:42020'
r = requests.get(f'{BASE}/api/health', timeout=5)
r.raise_for_status()
print(r.status_code)
print(r.json())
print(r.headers.get('X-Lab-App'))
print(r.headers.get('Set-Cookie'))
PY

200
{'status': 'ok', 'path': '/api/health'}
python-pentest-lab
lab_session=demo; Path=/

Send custom headers the way a web assessment often needs:

python3 - <<'PY'
import requests
r = requests.get(
    'http://127.0.0.1:42020/api/headers',
    headers={'User-Agent': 'lab-check/1.0', 'X-Token': 'demo'},
    timeout=5,
)
h = r.json()['headers']
print(h.get('User-Agent'))
print(h.get('X-Token'))
PY

lab-check/1.0
demo

POST JSON and attach a cookie:

python3 - <<'PY'
import requests
r = requests.post(
    'http://127.0.0.1:42020/api/echo',
    json={'host': 'lab', 'check': 'alive'},
    cookies={'lab_session': 'demo'},
    timeout=5,
)
print(r.status_code)
print(r.json())
PY

200
{'echo': {'host': 'lab', 'check': 'alive'}, 'cookie': 'lab_session=demo'}

Habits that matter on real jobs:

• Always pass timeout=
• Call r.raise_for_status() when a non-2xx should abort the script
• Do not use verify=False outside a closed lab unless you are deliberately testing certificate-validation behavior
• Use a requests.Session() when cookies or auth headers repeat

For directory and parameter discovery at scale, stay with purpose-built fuzzers such as ffuf and use Python to orchestrate or parse results.

Work with TCP sockets in Python

Sockets teach what HTTP libraries hide: connect, send, receive, close.

python3 - <<'PY'
import socket
with socket.create_connection(('127.0.0.1', 42021), timeout=3) as s:
    s.sendall(b'ping-lab')
    print(s.recv(1024).decode())
PY

ECHO:ping-lab

That confirms the echo service is reachable. Use sockets for banner grabs, custom binary protocols, or teaching. Do not treat them as a replacement for mature scanners. For service version clues, see banner grabbing.

UDP follows the same idea with socket.SOCK_DGRAM, but remember UDP has no connection handshake, so “no reply” is ambiguous.

Parse Nmap XML with Python

Writing another SYN scanner is a common beginner trap. On engagements you already have Nmap (or Masscan) output. Parse it.

Capture XML while the lab listeners are up:

nmap -oX data/localhost-scan.xml -p 42020,42021,3333 127.0.0.1

The scan should finish in under a second on localhost. Then convert open ports to JSON:

python3 - <<'PY'
import json
import xml.etree.ElementTree as ET

root = ET.parse('data/localhost-scan.xml').getroot()
rows = []
for host in root.findall('host'):
    addr_el = host.find("address[@addrtype='ipv4']")
    addr = addr_el.get('addr') if addr_el is not None else None
    for port in host.findall('ports/port'):
        state = port.find('state').get('state')
        svc = port.find('service')
        rows.append({
            'addr': addr,
            'port': int(port.get('portid')),
            'proto': port.get('protocol'),
            'state': state,
            'service': svc.get('name') if svc is not None else None,
        })
print(json.dumps(rows, indent=2))
PY

[
  {
    "addr": "127.0.0.1",
    "port": 3333,
    "proto": "tcp",
    "state": "open",
    "service": "dec-notes"
  },
  {
    "addr": "127.0.0.1",
    "port": 42020,
    "proto": "tcp",
    "state": "open",
    "service": null
  },
  {
    "addr": "127.0.0.1",
    "port": 42021,
    "proto": "tcp",
    "state": "open",
    "service": null
  }
]

Pipe that JSON into later HTTP checks or a report. The same idea works for tool JSON (-oJ) when the scanner supports it.

Use Scapy for packet-level work

Scapy is the Python toolkit for building and dissecting packets. Keep deep crafting, sniffing, and protocol fuzzing in a dedicated Scapy lesson. Here you only need the mental model.

Craft a TCP SYN toward the lab HTTP port without sending it:

python3 - <<'PY'
from scapy.all import IP, TCP
pkt = IP(dst='127.0.0.1') / TCP(dport=42020, flags='S')
print(pkt.summary())
PY

IP / TCP 127.0.0.1:ftp_data > 127.0.0.1:42020 S

Scapy may render well-known port numbers as service names. ftp_data here is only the auto-selected source port label, not an FTP connection. Call pkt.show() when you want the full field dump.

Sending crafted packets often needs elevated privileges and clear authorization. Prefer Nmap for routine port discovery; reach for Scapy when you must control flags, payloads, or unusual stacks.

Treat Impacket as a Python toolkit

Impacket is a Python library plus many CLI scripts for Windows and directory protocols (SMB, MSRPC, Kerberos helpers, and more). On Kali you usually install python3-impacket and impacket-scripts.

Confirm both pieces are present:

dpkg-query -W python3-impacket impacket-scripts

impacket-scripts	1.10
python3-impacket	0.13.0+git20251120-9c2d8b6-0kali1

Confirm the library import path as well:

python3 -c 'import impacket; print(impacket.__path__[0])'

/usr/lib/python3/dist-packages/impacket

What that means for scripting:

• Import protocol classes when you build custom automation
• Call packaged scripts (impacket-wmiexec, impacket-secretsdump, and siblings) from subprocess when the CLI already solves the task
• Keep secrets in env vars or a secrets file your RoE allows. Never commit them

This article does not walk lateral-movement recipes. For authorized SMB/WinRM execution and Impacket shells on an AD lab, use the NetExec and Impacket lesson.

Add concurrency for authorized enumeration

When you check many independent URLs, threads (or asyncio) cut wall-clock time. Keep the framing honest: you are parallelizing authorized checks, not “brute-forcing faster.”

Rules of thumb:

• Cap workers (--workers 4 is a sane lab default)
• Keep per-target timeouts
• Bound the target list to in-scope hosts
• Prefer concurrent.futures.ThreadPoolExecutor for blocking I/O (requests, sockets)

The service-check tool below uses a small thread pool. Bump workers only when the RoE and the target can absorb the load.

Build a small security automation tool

Put the pieces together: JSON targets in, JSON results out, timeouts, logging, and limited concurrency.

In this tool, ok: true means the check completed without a transport failure or HTTP 5xx. A 404 or 401 still counts as ok because the service answered. It does not mean the endpoint returned the business result you wanted.

Save a compact scripts/service_check.py (HTTP, TCP, and optional TLS):

#!/usr/bin/env python3
"""Authorized lab helper: HTTP reachability, TCP connect, optional TLS."""
from __future__ import annotations

import argparse
import json
import logging
import socket
import ssl
import sys
from concurrent.futures import ThreadPoolExecutor, as_completed
from typing import Any

import requests

log = logging.getLogger('service_check')


def check_http(url: str, timeout: float) -> dict[str, Any]:
    try:
        r = requests.get(url, timeout=timeout, allow_redirects=True)
        return {
            'target': url,
            'ok': r.status_code < 500,
            'detail': {
                'type': 'http',
                'status': r.status_code,
                'server': r.headers.get('Server'),
                'x_lab_app': r.headers.get('X-Lab-App'),
            },
        }
    except requests.RequestException as exc:
        return {'target': url, 'ok': False, 'detail': {'type': 'http', 'error': str(exc)}}


def check_tcp(host: str, port: int, timeout: float) -> dict[str, Any]:
    target = f'{host}:{port}'
    try:
        with socket.create_connection((host, port), timeout=timeout):
            return {'target': target, 'ok': True, 'detail': {'type': 'tcp', 'state': 'open'}}
    except OSError as exc:
        return {'target': target, 'ok': False, 'detail': {'type': 'tcp', 'error': str(exc)}}


def check_tls(host: str, port: int, timeout: float, insecure: bool = False) -> dict[str, Any]:
    target = f'{host}:{port}'
    try:
        ctx = ssl._create_unverified_context() if insecure else ssl.create_default_context()
        with socket.create_connection((host, port), timeout=timeout) as sock:
            with ctx.wrap_socket(sock, server_hostname=host) as ssock:
                return {
                    'target': target,
                    'ok': True,
                    'detail': {'type': 'tls', 'version': ssock.version(), 'insecure': insecure},
                }
    except (OSError, ssl.SSLError) as exc:
        return {'target': target, 'ok': False, 'detail': {'type': 'tls', 'error': str(exc)}}


def run_one(item: dict[str, Any], timeout: float) -> dict[str, Any]:
    kind = item.get('type')
    if kind == 'http':
        return check_http(item['url'], timeout)
    if kind == 'tcp':
        return check_tcp(item['host'], int(item['port']), timeout)
    if kind == 'tls':
        return check_tls(
            item['host'], int(item['port']), timeout, insecure=bool(item.get('insecure'))
        )
    return {'target': str(item), 'ok': False, 'detail': {'error': f'unknown type {kind}'}}


def main() -> int:
    p = argparse.ArgumentParser(description='Check authorized HTTP/TCP/TLS targets')
    p.add_argument('-t', '--targets', required=True)
    p.add_argument('-o', '--output')
    p.add_argument('--timeout', type=float, default=3.0)
    p.add_argument('--workers', type=int, default=4)
    p.add_argument('-v', '--verbose', action='store_true')
    args = p.parse_args()

    logging.basicConfig(
        level=logging.DEBUG if args.verbose else logging.INFO,
        format='%(levelname)s %(message)s',
    )

    with open(args.targets, encoding='utf-8') as fh:
        targets = json.load(fh)

    results: list[dict[str, Any]] = []
    with ThreadPoolExecutor(max_workers=max(1, args.workers)) as pool:
        futs = [pool.submit(run_one, item, args.timeout) for item in targets]
        for fut in as_completed(futs):
            res = fut.result()
            results.append(res)
            log.info('%s ok=%s detail=%s', res['target'], res['ok'], res['detail'])

    payload = sorted(results, key=lambda r: r['target'])
    text = json.dumps(payload, indent=2)
    if args.output:
        with open(args.output, 'w', encoding='utf-8') as fh:
            fh.write(text + '\n')
        log.info('wrote %s', args.output)
    else:
        print(text)
    return 0 if all(r['ok'] for r in results) else 1


if __name__ == '__main__':
    sys.exit(main())

Create data/targets.json (adjust the TLS row if Gophish is not listening on 3333):

[
  {"type": "http", "url": "http://127.0.0.1:42020/api/health"},
  {"type": "tcp", "host": "127.0.0.1", "port": 42021},
  {"type": "tls", "host": "127.0.0.1", "port": 3333, "insecure": true}
]

insecure: true is only for lab self-signed certs. On client work, fix trust stores or document the finding instead. For verified TLS checks, use the DNS name that appears in the certificate rather than substituting an IP address.

Run the tool:

python3 scripts/service_check.py -t data/targets.json -o out/results.json --workers 3

INFO 127.0.0.1:42021 ok=True detail={'type': 'tcp', 'state': 'open'}
INFO http://127.0.0.1:42020/api/health ok=True detail={'type': 'http', 'status': 200, 'server': 'BaseHTTP/0.6 Python/3.13.12', 'x_lab_app': 'python-pentest-lab'}
INFO 127.0.0.1:3333 ok=True detail={'type': 'tls', 'version': 'TLSv1.3', 'insecure': True}
INFO wrote out/results.json

Order of INFO lines can vary because workers finish out of order. The JSON file is sorted by target:

cat out/results.json

[
  {
    "target": "127.0.0.1:3333",
    "ok": true,
    "detail": {
      "type": "tls",
      "version": "TLSv1.3",
      "insecure": true
    }
  },
  {
    "target": "127.0.0.1:42021",
    "ok": true,
    "detail": {
      "type": "tcp",
      "state": "open"
    }
  },
  {
    "target": "http://127.0.0.1:42020/api/health",
    "ok": true,
    "detail": {
      "type": "http",
      "status": 200,
      "server": "BaseHTTP/0.6 Python/3.13.12",
      "x_lab_app": "python-pentest-lab"
    }
  }
]

Exit status is 0 only when every check reports ok: true. That makes the script usable in CI-style lab gates.

Handle errors, timeouts, and logging

Quality markers that separate lab toys from engagement scripts:

• Timeouts on every network call
• Caught exceptions turned into structured ok: false rows (or a non-zero exit)
• logging instead of only print
• Narrow except clauses (requests.RequestException, OSError, ssl.SSLError) instead of bare except:
• Clear CLI flags via argparse

When something fails, keep the error string in the JSON detail so the report explains why, not only that a check failed.

Package and reuse Python security scripts

For anything you will run twice:

python3 -m venv .venv

If venv is missing on Kali, install it with sudo apt install python3-venv, then retry.

Activate and declare dependencies (Kali system packages are fine for this lesson; use a venv when you need isolated libs):

printf '%s\n' 'requests>=2.32' > requirements.txt

Document the entry point:

python3 scripts/service_check.py --help

usage: service_check.py [-h] -t TARGETS [-o OUTPUT] [--timeout TIMEOUT]
                        [--workers WORKERS] [-v]

Check authorized HTTP/TCP/TLS targets

options:
  -h, --help            show this help message and exit
  -t, --targets TARGETS
  -o, --output OUTPUT
  --timeout TIMEOUT
  --workers WORKERS
  -v, --verbose

Ship the script, requirements.txt, and a sample targets.json with secrets removed.

Common Python pentest scripting mistakes

Troubleshoot Python pentest scripts

Summary

Python earns its place on a pentest laptop when it reads approved targets, drives or parses real tools, talks HTTP and sockets with timeouts, and writes JSON you can drop into a report. The lab path here (local HTTP and TCP services, requests, socket echo, Nmap XML parsing, a Scapy packet preview, and a small concurrent service-check tool) is the reusable pattern.

Leave packet crafting depth to Scapy-focused material, and leave Windows protocol execution to Impacket/NetExec lessons. Keep concurrency capped, keep secrets out of source, and treat verify=False as a lab exception you document.

Next, wire this checker to live in-scope URLs from your RoE, or extend the JSON schema with banner and TLS certificate fields your reporting template already expects.

References

• Python 3 documentation
• Requests documentation
• Scapy documentation
• Impacket
• Nmap XML output