Subprocess & OS Automation
Run shell commands from Python using subprocess, automate system tasks with the os and pathlib modules, and build security automation scripts.
Learning Objectives
- → Run shell commands with subprocess.run() and capture output
- → Use subprocess.Popen for interactive and streaming processes
- → Navigate the filesystem with os and pathlib
- → Automate common security tasks: nmap wrapping, file hashing
- → Handle command injection risks safely
subprocess.run()
The modern way to run external commands:
import subprocess
# Run a command, capture output
result = subprocess.run(
["ls", "-la", "/tmp"], # list form — NEVER shell=True with user input
capture_output=True,
text=True, # decode bytes to str
timeout=10,
)
print(result.stdout)
print(result.returncode) # 0 = success
if result.returncode != 0:
print("Error:", result.stderr)
NEVER Use shell=True with User Input
# DANGEROUS — command injection
user_input = "example.com; rm -rf /"
subprocess.run(f"ping {user_input}", shell=True) # DON'T
# SAFE — list form passes args directly, no shell interpretation
subprocess.run(["ping", "-c", "1", user_input])
Wrapping nmap
import subprocess, json
def nmap_scan(host, flags="-sV -T4 --open"):
cmd = ["nmap"] + flags.split() + ["-oX", "-", host]
result = subprocess.run(cmd, capture_output=True, text=True, timeout=60)
return result.stdout # XML output
output = nmap_scan("127.0.0.1", "-p 22,80,443")
print(output[:500])
Popen for Streaming Output
import subprocess
proc = subprocess.Popen(
["ping", "-c", "5", "8.8.8.8"],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
)
for line in proc.stdout:
print(line, end="")
proc.wait()
print("Return code:", proc.returncode)
os Module
import os
# Paths
cwd = os.getcwd()
home = os.path.expanduser("~")
joined = os.path.join(home, "tools", "scanner.py")
# File info
size = os.path.getsize("scan.log")
exists = os.path.exists("config.json")
# Directory operations
os.makedirs("reports/2024", exist_ok=True)
# Environment variables
path = os.environ.get("PATH", "")
api_key = os.environ.get("API_KEY")
pathlib (Modern Alternative)
from pathlib import Path
base = Path.home() / "cybertools"
base.mkdir(exist_ok=True)
for f in base.glob("*.json"):
print(f.name, f.stat().st_size)
log = Path("auth.log")
if log.exists():
lines = log.read_text().splitlines()
File Hashing
import hashlib
def sha256_file(path):
h = hashlib.sha256()
with open(path, "rb") as f:
for chunk in iter(lambda: f.read(8192), b""):
h.update(chunk)
return h.hexdigest()
print(sha256_file("/bin/ls"))
Mini Project – System Recon Script
import subprocess, platform, socket, json
def system_recon():
info = {
"hostname": socket.gethostname(),
"platform": platform.system(),
"release": platform.release(),
"arch": platform.machine(),
}
# Open ports (requires ss or netstat)
r = subprocess.run(["ss", "-tlnp"], capture_output=True, text=True)
info["listening_ports"] = r.stdout
# Running processes count
r2 = subprocess.run(["ps", "-e"], capture_output=True, text=True)
info["process_count"] = len(r2.stdout.strip().splitlines()) - 1
return info
data = system_recon()
print(json.dumps({k: v[:100] if isinstance(v, str) else v for k, v in data.items()}, indent=2))
Use subprocess.run to run uname -a and whoami. Capture and print their output. Check the return code for each.
What argument to subprocess.run() captures stdout and stderr?
What returncode indicates success?
Implement sha256_file() from the lesson. Hash 3 different files and store the results in a dict {filename: hash}. Write the dict to a JSON file as a simple integrity baseline.
What hashlib function creates a SHA-256 hasher?
Why read the file in chunks ('rb', 8192 bytes) instead of all at once?
Run the system_recon() function. Extend it to also capture disk usage (df -h) and current user's groups (groups). Write the full report to report.json.
Why should you pass commands as a list instead of a string with shell=True?