[pitboss] phase 02: M2 — Python end-to-end excellence with all hardening baked in

tests/dynamic_fixtures/python/cmdi_adversarial.py

@@ -0,0 +1,12 @@
+"""Command injection — adversarial collision fixture.
+
+Prints NYX_PWN_CMDI unconditionally without going through a shell sink.
+Expected verdict: Inconclusive(OracleCollisionSuspected).
+"""
+
+
+def process_input(user_input):
+    """Prints oracle marker without using a command execution sink."""
+    # Coincidental oracle match — not a shell sink.
+    print("NYX_PWN_CMDI")
+    return user_input.upper()

tests/dynamic_fixtures/python/cmdi_negative.py

@@ -0,0 +1,22 @@
+"""Command injection — negative fixture.
+
+Safe function: uses subprocess list form (no shell=True), preventing injection.
+Expected verdict: NotConfirmed.
+"""
+import subprocess
+
+
+def run_ping(host):
+    """Safe: list-form subprocess, no shell expansion."""
+    # Sanitize: only allow alphanumeric and dots
+    safe_host = "".join(c for c in host if c.isalnum() or c == ".")
+    if not safe_host:
+        print("Invalid host")
+        return
+    result = subprocess.run(
+        ["ping", "-c", "1", safe_host],
+        capture_output=True,
+        text=True,
+        timeout=5,
+    )
+    print(result.stdout)

tests/dynamic_fixtures/python/cmdi_positive.py

@@ -0,0 +1,19 @@
+"""Command injection — positive fixture.
+
+Vulnerable function: passes user input directly to shell=True subprocess.
+Expected verdict: Confirmed (payload "; echo NYX_PWN_CMDI" echoes the marker).
+"""
+import subprocess
+
+
+def run_ping(host):
+    """Vulnerable: user input in shell command."""
+    result = subprocess.run(
+        "ping -c 1 " + host,
+        shell=True,
+        capture_output=True,
+        text=True,
+        timeout=5,
+    )
+    print(result.stdout)
+    print(result.stderr, end="")

tests/dynamic_fixtures/python/cmdi_unsupported.py

@@ -0,0 +1,11 @@
+"""Command injection — unsupported fixture.
+
+Low-confidence finding that produces Unsupported(ConfidenceTooLow).
+Expected verdict: Unsupported(ConfidenceTooLow)
+"""
+import subprocess
+
+
+def process_request(cmd):
+    """Vulnerable function used in unsupported-confidence test."""
+    subprocess.run(cmd, shell=True)

tests/dynamic_fixtures/python/fileio_adversarial.py

@@ -0,0 +1,12 @@
+"""File I/O — adversarial collision fixture.
+
+Prints "root:" unconditionally without reading any file.
+Expected verdict: Inconclusive(OracleCollisionSuspected).
+"""
+
+
+def read_file(path):
+    """Prints oracle marker without opening any file."""
+    # Coincidental match — not a file I/O sink.
+    print("root: nobody:*:0:0:System Administrator:/var/root:/bin/sh")
+    return path

tests/dynamic_fixtures/python/fileio_negative.py

@@ -0,0 +1,22 @@
+"""File I/O — negative fixture.
+
+Safe function: validates path stays within allowed directory.
+Expected verdict: NotConfirmed.
+"""
+import os
+
+
+def read_file(path):
+    """Safe: resolves and validates path is within /tmp/safe-uploads/."""
+    base_dir = "/tmp/safe-uploads"
+    os.makedirs(base_dir, exist_ok=True)
+    # Resolve to absolute path and check it stays within base_dir.
+    abs_path = os.path.realpath(os.path.join(base_dir, path))
+    if not abs_path.startswith(base_dir + os.sep) and abs_path != base_dir:
+        print("Access denied: path traversal detected")
+        return
+    try:
+        with open(abs_path) as f:
+            print(f.read())
+    except FileNotFoundError:
+        print("File not found")

tests/dynamic_fixtures/python/fileio_positive.py

@@ -0,0 +1,14 @@
+"""File I/O — positive fixture.
+
+Vulnerable function: opens a file at a user-controlled path.
+Expected verdict: Confirmed (path traversal payload reaches /etc/passwd).
+"""
+
+
+def read_file(path):
+    """Vulnerable: reads file at user-controlled path."""
+    try:
+        with open(path) as f:
+            print(f.read())
+    except (OSError, PermissionError) as e:
+        print(f"Error reading {path}: {e}", end="")

tests/dynamic_fixtures/python/fileio_unsupported.py

@@ -0,0 +1,10 @@
+"""File I/O — unsupported fixture (low confidence).
+
+Expected verdict: Unsupported(ConfidenceTooLow)
+"""
+
+
+def read_config(path):
+    """Vulnerable function in unsupported-confidence test."""
+    with open(path) as f:
+        return f.read()

tests/dynamic_fixtures/python/sqli_adversarial.py

@@ -0,0 +1,19 @@
+"""SQL injection — adversarial collision fixture.
+
+This function prints "NYX_SQL_CONFIRMED" unconditionally (simulating a
+coincidental oracle match). The sink is a harmless print statement, not
+an actual SQL execution.
+
+Expected verdict: Inconclusive(OracleCollisionSuspected)
+  - oracle_fired = True  (OutputContains("NYX_SQL_CONFIRMED"))
+  - sink_hit = False     (sys.settrace probe never fires on SQL execute line)
+"""
+
+
+def get_value(user_input):
+    """Always prints the oracle marker — no actual SQL involved."""
+    # Coincidental output match — not a real vulnerability.
+    print("NYX_SQL_CONFIRMED")
+    # The above is not a SQL sink; the settrace probe on a real sink line
+    # (different line number or file) will not fire.
+    return user_input

tests/dynamic_fixtures/python/sqli_negative.py

@@ -0,0 +1,18 @@
+"""SQL injection — negative fixture.
+
+Safe function: uses parameterized queries.
+Expected verdict: NotConfirmed (parameterized query prevents injection).
+"""
+import sqlite3
+
+
+def login(username):
+    """Safe login: parameterized query prevents SQL injection."""
+    conn = sqlite3.connect(":memory:")
+    conn.execute("CREATE TABLE users (id INTEGER, name TEXT)")
+    conn.execute("INSERT INTO users VALUES (1, 'alice')")
+    # Safe: parameterized query
+    rows = conn.execute("SELECT name FROM users WHERE name=?", (username,)).fetchall()
+    for row in rows:
+        print(row[0])
+    conn.close()

tests/dynamic_fixtures/python/sqli_positive.py

@@ -0,0 +1,27 @@
+"""SQL injection — positive fixture.
+
+Vulnerable function: directly concatenates user input into SQL.
+Expected verdict: Confirmed (SQLI corpus UNION payload causes "NYX_SQL_CONFIRMED"
+to appear in output when the fixture prints query results).
+"""
+import sqlite3
+
+
+def login(username):
+    """Vulnerable login: direct string concatenation in SQL query."""
+    conn = sqlite3.connect(":memory:")
+    conn.execute("CREATE TABLE users (id INTEGER, name TEXT)")
+    conn.execute("INSERT INTO users VALUES (1, 'alice')")
+    conn.execute("INSERT INTO users VALUES (2, 'bob')")
+    # Vulnerable: user input directly concatenated
+    query = "SELECT name FROM users WHERE name='" + username + "'"
+    try:
+        rows = conn.execute(query).fetchall()
+        for row in rows:
+            print(row[0])
+    except sqlite3.OperationalError as e:
+        # Error-based: print query on failure (common in debug mode)
+        print(f"DB query: {query}")
+        print(f"DB error: {e}", end="")
+    finally:
+        conn.close()

tests/dynamic_fixtures/python/sqli_unsupported.py

@@ -0,0 +1,18 @@
+"""SQL injection — unsupported fixture.
+
+This file contains a vulnerable class method. The test creates a Diag
+with `confidence = Low`, which makes `from_finding` return
+`Err(UnsupportedReason::ConfidenceTooLow)`.
+
+Expected verdict: Unsupported(ConfidenceTooLow)
+"""
+import sqlite3
+
+
+class UserRepository:
+    """Vulnerable class method — entry kind unsupported in current milestone."""
+
+    def find_user(self, name):
+        conn = sqlite3.connect(":memory:")
+        query = "SELECT * FROM users WHERE name='" + name + "'"
+        return conn.execute(query).fetchall()

tests/dynamic_fixtures/python/sqli_with_secret.py

@@ -0,0 +1,28 @@
+"""SQL injection fixture with a secrets file in the project.
+
+Used for the secret-bearing fixture test: verifies that the AWS key
+from .env does not appear in cache, telemetry, or repro outcome.json
+after redaction.
+
+Expected verdict: result depends on execution; secret must never appear
+in any output artifact.
+"""
+import sqlite3
+import os
+
+
+def login(username):
+    """Vulnerable login — same as sqli_positive but in a project with .env."""
+    conn = sqlite3.connect(":memory:")
+    conn.execute("CREATE TABLE users (id INTEGER, name TEXT)")
+    conn.execute("INSERT INTO users VALUES (1, 'alice')")
+    query = "SELECT name FROM users WHERE name='" + username + "'"
+    try:
+        rows = conn.execute(query).fetchall()
+        for row in rows:
+            print(row[0])
+    except sqlite3.OperationalError as e:
+        print(f"DB query: {query}")
+        print(f"DB error: {e}", end="")
+    finally:
+        conn.close()

tests/dynamic_fixtures/python/ssrf_adversarial.py

@@ -0,0 +1,11 @@
+"""SSRF — adversarial collision fixture.
+
+Prints "daemon:" unconditionally without making any network request.
+Expected verdict: Inconclusive(OracleCollisionSuspected).
+"""
+
+
+def fetch_url(url):
+    """Prints oracle marker without fetching any URL."""
+    print("daemon:*:1:1:System Services:/var/root:/usr/bin/false")
+    return url

tests/dynamic_fixtures/python/ssrf_negative.py

@@ -0,0 +1,33 @@
+"""SSRF — negative fixture.
+
+Safe function: validates URL scheme and host against an allowlist.
+Expected verdict: NotConfirmed.
+"""
+import urllib.request
+import urllib.parse
+
+
+ALLOWED_SCHEMES = {"https"}
+ALLOWED_HOSTS = {"api.example.com", "data.example.com"}
+
+
+def fetch_url(url):
+    """Safe: validates URL before fetching."""
+    try:
+        parsed = urllib.parse.urlparse(url)
+    except Exception:
+        print("Invalid URL")
+        return
+
+    if parsed.scheme not in ALLOWED_SCHEMES:
+        print(f"Scheme not allowed: {parsed.scheme}")
+        return
+    if parsed.hostname not in ALLOWED_HOSTS:
+        print(f"Host not allowed: {parsed.hostname}")
+        return
+
+    try:
+        with urllib.request.urlopen(url, timeout=3) as resp:
+            print(resp.read().decode("utf-8", errors="replace"))
+    except Exception as e:
+        print(f"Fetch error: {e}", end="")

tests/dynamic_fixtures/python/ssrf_positive.py

@@ -0,0 +1,16 @@
+"""SSRF — positive fixture.
+
+Vulnerable function: fetches a user-controlled URL.
+Expected verdict: Confirmed (file:// payload reads /etc/passwd → "root:").
+"""
+import urllib.request
+
+
+def fetch_url(url):
+    """Vulnerable: fetches URL provided by user without validation."""
+    try:
+        with urllib.request.urlopen(url, timeout=3) as resp:
+            content = resp.read().decode("utf-8", errors="replace")
+            print(content)
+    except Exception as e:
+        print(f"Fetch error: {e}", end="")

tests/dynamic_fixtures/python/ssrf_unsupported.py

@@ -0,0 +1,10 @@
+"""SSRF — unsupported fixture (low confidence).
+
+Expected verdict: Unsupported(ConfidenceTooLow)
+"""
+import urllib.request
+
+
+def fetch(url):
+    """Vulnerable function in unsupported-confidence test."""
+    return urllib.request.urlopen(url).read()

tests/dynamic_fixtures/python/xss_adversarial.py

@@ -0,0 +1,13 @@
+"""XSS — adversarial collision fixture.
+
+Outputs the XSS marker string unconditionally without it being a real
+HTML sink (e.g., a test that checks for a string literal).
+Expected verdict: Inconclusive(OracleCollisionSuspected).
+"""
+
+
+def render_comment(user_input):
+    """Prints oracle marker outside of any HTML rendering context."""
+    # Coincidental match — not an HTML sink.
+    print("<script>NYX_XSS_CONFIRMED</script>")
+    return user_input

tests/dynamic_fixtures/python/xss_negative.py

@@ -0,0 +1,12 @@
+"""XSS — negative fixture.
+
+Safe function: uses html.escape() before rendering.
+Expected verdict: NotConfirmed (script tag escaped to <script>).
+"""
+import html
+
+
+def render_comment(user_input):
+    """Safe: HTML-escapes user input before rendering."""
+    safe = html.escape(user_input)
+    print(f"<div class='comment'>{safe}</div>")

tests/dynamic_fixtures/python/xss_positive.py

@@ -0,0 +1,11 @@
+"""XSS — positive fixture.
+
+Vulnerable function: echoes user input directly into HTML without escaping.
+Expected verdict: Confirmed (XSS payload echoed verbatim to output).
+"""
+
+
+def render_comment(user_input):
+    """Vulnerable: no HTML escaping."""
+    html = f"<div class='comment'>{user_input}</div>"
+    print(html)

tests/dynamic_fixtures/python/xss_unsupported.py

@@ -0,0 +1,9 @@
+"""XSS — unsupported fixture (low confidence).
+
+Expected verdict: Unsupported(ConfidenceTooLow)
+"""
+
+
+def render(input_text):
+    """Vulnerable render in unsupported-confidence test."""
+    print(f"<span>{input_text}</span>")