Python for Security Engineering

# Run a single exercise
python strings/level_01_reverse_string.py

# Run all exercises in a topic
for f in strings/level_*.py; do python "$f" && echo "PASS: $f" || echo "FAIL: $f"; done

# Run everything
for f in */level_*.py; do python "$f" && echo "PASS: $f" || echo "FAIL: $f"; done

import re
from collections import Counter

LOG_LINE = r'Failed password for (\w+) from ([\d.]+) port (\d+)'

def parse_auth_failures(log_text: str) -> list[dict]:
    return [
        {"user": m.group(1), "ip": m.group(2), "port": int(m.group(3))}
        for m in re.finditer(LOG_LINE, log_text)
    ]

def find_brute_force(failures: list[dict], threshold: int = 5) -> list[str]:
    counts = Counter(f["ip"] for f in failures)
    return [ip for ip, count in counts.items() if count >= threshold]

import socket
from concurrent.futures import ThreadPoolExecutor

def scan_port(host: str, port: int, timeout: float = 0.5) -> int | None:
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
        s.settimeout(timeout)
        return port if s.connect_ex((host, port)) == 0 else None

def scan_ports(host: str, ports: range) -> list[int]:
    with ThreadPoolExecutor(max_workers=100) as pool:
        results = pool.map(lambda p: scan_port(host, p), ports)
    return [p for p in results if p is not None]

import hmac, hashlib

def sign_payload(payload: bytes, key: bytes) -> str:
    return hmac.new(key, payload, hashlib.sha256).hexdigest()

def verify_payload(payload: bytes, key: bytes, signature: str) -> bool:
    expected = sign_payload(payload, key)
    return hmac.compare_digest(expected, signature)  # constant-time — prevents timing attacks

import hmac

def safe_compare(a: str, b: str) -> bool:
    return hmac.compare_digest(a.encode(), b.encode())

# Never use == for secrets: early exit leaks information via timing

import math
from collections import Counter

def shannon_entropy(s: str) -> float:
    counts = Counter(s)
    length = len(s)
    return -sum((c / length) * math.log2(c / length) for c in counts.values())

def detect_dns_exfil(queries: list[str], threshold: float = 4.0) -> list[str]:
    suspicious = []
    for query in queries:
        subdomain = query.split('.')[0]
        if len(subdomain) > 20 and shannon_entropy(subdomain) > threshold:
            suspicious.append(query)
    return suspicious

# High entropy + long subdomain = likely base64/hex encoded data in DNS query

import re
from dataclasses import dataclass

@dataclass
class Rule:
    name: str
    patterns: list[bytes]

    def matches(self, content: bytes) -> bool:
        return all(re.search(p, content) for p in self.patterns)

RULES = [
    Rule("crypto_miner", [rb"stratum\+tcp", rb"xmrig"]),
    Rule("reverse_shell", [rb"/bin/sh", rb"connect\(", rb"socket\("]),
    Rule("credential_dump", [rb"lsass", rb"mimikatz", rb"sekurlsa"]),
]

def scan_binary(path: str) -> list[str]:
    with open(path, 'rb') as f:
        content = f.read()
    return [rule.name for rule in RULES if rule.matches(content)]

from collections import deque
import time

class RateLimiter:
    """Sliding window rate limiter. O(1) amortised allow() calls."""

    def __init__(self, max_requests: int, window_seconds: float):
        self.max_requests = max_requests
        self.window = window_seconds
        self.timestamps: deque[float] = deque()

    def allow(self) -> bool:
        now = time.monotonic()
        while self.timestamps and now - self.timestamps[0] > self.window:
            self.timestamps.popleft()
        if len(self.timestamps) < self.max_requests:
            self.timestamps.append(now)
            return True
        return False

# Usage: limiter = RateLimiter(max_requests=100, window_seconds=60)

# Cryptography & integrity
import hashlib          # hashlib.sha256(data).hexdigest()
import hmac             # hmac.new(key, msg, hashlib.sha256).hexdigest()
import secrets          # secrets.token_hex(32), secrets.token_urlsafe(32)

# Encoding
import base64           # base64.b64encode/b64decode
import binascii         # hexlify / unhexlify

# Network
import socket           # TCP/UDP, connect_ex for port scanning
import ssl              # TLS wrapping, ssl.create_default_context()
import ipaddress        # ip_address, ip_network, CIDR membership

# Process & OS
import subprocess       # run(args, shell=False, capture_output=True)
import os               # os.environ, os.walk, os.stat
import pathlib          # Path traversal, iterdir, glob

# Parsing
import re               # findall, finditer, sub, compile
import json             # loads / dumps
import csv              # DictReader for structured log files
import ast              # ast.literal_eval — safe eval for Python literals

# Data structures
from collections import Counter, defaultdict, deque, OrderedDict
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor

# Time
import time             # time.monotonic() for rate limiters (no clock drift)
import datetime         # datetime.utcnow(), timedelta