Python für Security-Automation.

Subprocess-Disziplin

• Niemals shell=True mit User-Input. Shell-Injection trivial. List-Form nutzen: subprocess.run(["nmap", "-sS", target]).
• Beide Streams capturen. capture_output=True + text=True. Tools schreiben Progress nach stderr; ignorieren verliert die Hälfte des Signals.
• Encoding. encoding="utf-8", errors="replace". Echter Tool-Output enthält binären Müll; Default-Strict-Mode raised auf jedes komische Byte.
• Timeout. Immer timeout=300. Hängender Subprocess in 10-Thread-Scanner = wedged Scanner. subprocess.TimeoutExpired catchen, weitermachen.
• Langer Output. Streamen via Popen mit stdout=subprocess.PIPE und zeilenweise lesen. capture_output buffert alles im RAM und OOMt bei großen nmap-Läufen.

Robustes HTTP

import httpx
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(
    stop=stop_after_attempt(5),
    wait=wait_exponential(multiplier=1, min=1, max=30),
)
async def fetch(client, url):
    r = await client.get(url, timeout=15.0, follow_redirects=False)
    if r.status_code == 429:
        retry_after = int(r.headers.get("retry-after", "5"))
        await asyncio.sleep(retry_after)
        r.raise_for_status()
    return r

• httpx statt requests. Native async, HTTP/2, sane Timeouts.
• follow_redirects=False per default. SSRF-Probes und Login-Flows wollen Redirects inspizieren, nicht jagen.
• Retry-After honorieren. Ignorieren beschleunigt Rate-Limit; Honorieren kommt durch.
• Per-Request-Timeout. Connect + read + write + pool. Tuple-Form: timeout=httpx.Timeout(5.0, read=10.0).

Async-Scanner-Muster

import asyncio
import httpx

async def scan(targets, concurrency=50):
    sem = asyncio.Semaphore(concurrency)
    async with httpx.AsyncClient() as client:
        async def one(t):
            async with sem:
                try:
                    return await fetch(client, t)
                except Exception as e:
                    return ("error", t, str(e))
        return await asyncio.gather(*[one(t) for t in targets])

• Semaphore für Concurrency-Cap. Ohne öffnet asyncio fröhlich 10000 Sockets und hängt den Event-Loop.
• Exceptions pro Task catchen. Sonst killt ein ConnectionRefused den ganzen gather.
• Einzelner AsyncClient. Connection-Pool wiederverwenden; per-Request-Client-Erstellung negiert Async-Nutzen.

Regex — Muster und ReDoS

• Nested Quantifiers vermeiden. (a+)+, (a|a)*, (a*)* — alle katastrophal. Linearer Input × exponentielle Zeit.
• Alternation mit überlappenden Prefixes vermeiden. (abc|ab|a)* — Backtracking explodiert. Anchoren oder Prefix herausziehen.
• Possessive Quantifier oder Atomic Groups nutzen wenn die Engine sie unterstützt — (?>...) in PCRE verhindert Backtracking in die Gruppe.
• Regex-Timeout setzen. Pythons re hat keinen Built-in-Timeout. re2 für untrusted Input (lineare Zeit garantiert).
• Häufige Security-Regexes.
  • Email loose: r"[\w.+-]+@[\w.-]+\.[a-z]{2,}" (case-insensitive).
  • IPv4: r"\b(?:\d{1,3}\.){3}\d{1,3}\b".
  • Bearer-Token: r"[Bb]earer\s+([A-Za-z0-9_\-\.=]+)".
  • UUIDv4: r"[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}".

Log-Triage — pandas-Idiome

import pandas as pd

df = pd.read_json("auth.log.json", lines=True)
suspicious = (
    df.assign(ts=pd.to_datetime(df.timestamp))
      .query("event_type == 'login_failed'")
      .groupby("source_ip")
      .agg(count=("ts","size"), first=("ts","min"), last=("ts","max"))
      .query("count > 50")
      .sort_values("count", ascending=False)
)