{"id":"postgres-query-with-params","version":"1.0.0","primitive":"code_execution","description":"PostgreSQL connection string e.g. \"postgresql://user:pass@host:5432/dbname\"","registry_refs":["psycopg2-binary"],"tags":[],"solves":[],"auth_required":true,"verified":false,"last_verified":"null","next_check":"2026-07-30","eval_result":"null","eval_env":"null","mast":[],"ref":"https://arxiv.org/abs/2503.13657","inputs":[{"name":"DATABASE_URL","required":true,"description":"PostgreSQL connection string e.g. \"postgresql://user:pass@host:5432/dbname\""}],"executable":"# ============================================\n# checklist:     postgres-query-with-params\n# version:       1.0.0\n# primitive:     code_execution\n# description:   Run a parameterized SELECT query against PostgreSQL avoiding f-string SQL injection footgun\n# registry_refs: psycopg2-binary\n# auth_required: true\n# verified:      false\n# last_verified: null\n# next_check:    2026-07-30\n# eval_result:   null\n# eval_env:      null\n#\n# inputs:\n#   - name: DATABASE_URL\n#     required: true\n#     description: PostgreSQL connection string e.g. \"postgresql://user:pass@host:5432/dbname\"\n#\n# OUTPUTS:\n#   rows_returned    — number of rows returned\n#   injection_blocked — true if SQL injection attempt was safely blocked\n#   parameterized_ok  — true if parameterized query returned correct results\n#\n# MAST FAILURE MODES ADDRESSED:\n# FM-1.1 Disobey Task Specification        — parameterized queries, never f-string SQL\n# FM-2.6 Reasoning-Action Mismatch         — demonstrates safe vs unsafe query patterns\n# FM-3.3 Incorrect Verification            — injection attempt verified as blocked\n#\n# ref: https://arxiv.org/abs/2503.13657\n# ============================================\n\nimport sys\nimport os\nimport subprocess\nimport time\nimport urllib.request\nimport json\n\n# ─────────────────────────────────────────\n# PRE_EXECUTION\n# ─────────────────────────────────────────\n\nfor attempt in range(2):\n    try:\n        req = urllib.request.Request(\n            \"https://checklist.day/api/registry/psycopg2-binary\",\n            headers={\"User-Agent\": \"checklist-agent/1.0\"}\n        )\n        with urllib.request.urlopen(req, timeout=10) as resp:\n            registry = json.loads(resp.read())\n            break\n    except Exception as e:\n        if attempt == 1:\n            print(f\"ABORT: registry unreachable — {e}\")\n            sys.exit(1)\n        time.sleep(2)\n\nwarnings = registry.get(\"warnings\", [])\nif warnings:\n    print(\"[psycopg2-binary] WARNINGS:\")\n    for w in warnings if isinstance(warnings, list) else [warnings]:\n        print(f\"  ⚠ {w}\")\n\n# ─────────────────────────────────────────\n# EXECUTION\n# ─────────────────────────────────────────\n\nsubprocess.check_call([sys.executable, \"-m\", \"pip\", \"install\", \"-q\", \"psycopg2-binary>=2.9.9\"])\n\nimport psycopg2\n\nDATABASE_URL = os.environ.get(\"DATABASE_URL\")\nif not DATABASE_URL:\n    print(\"ABORT: DATABASE_URL env var not set\")\n    sys.exit(1)\n\nif DATABASE_URL.startswith(\"postgres://\"):\n    DATABASE_URL = DATABASE_URL.replace(\"postgres://\", \"postgresql://\", 1)\n\nTABLE = \"checklist_query_test\"\nconn = psycopg2.connect(DATABASE_URL, connect_timeout=10)\n\ntry:\n    with conn.cursor() as cur:\n        # Setup\n        cur.execute(f\"\"\"\n            CREATE TABLE IF NOT EXISTS {TABLE} (\n                id    SERIAL PRIMARY KEY,\n                name  TEXT NOT NULL,\n                score INTEGER NOT NULL\n            )\n        \"\"\")\n        cur.execute(f\"DELETE FROM {TABLE}\")\n        cur.executemany(\n            f\"INSERT INTO {TABLE} (name, score) VALUES (%s, %s)\",\n            [(\"alice\", 90), (\"bob\", 75), (\"carol\", 88)]\n        )\n        conn.commit()\n\n        # 1. Safe parameterized query\n        # FOOTGUN: NEVER do f\"SELECT ... WHERE name = '{user_input}'\" — use %s\n        min_score = 80\n        cur.execute(f\"SELECT name, score FROM {TABLE} WHERE score >= %s ORDER BY score DESC\", (min_score,))\n        rows = cur.fetchall()\n        rows_returned = len(rows)\n        print(f\"  parameterized query returned {rows_returned} rows (score >= {min_score})\")\n\n        # 2. Demonstrate injection is blocked\n        # A malicious input that would break an f-string query\n        malicious_input = \"' OR '1'='1\"\n        cur.execute(f\"SELECT COUNT(*) FROM {TABLE} WHERE name = %s\", (malicious_input,))\n        injection_count = cur.fetchone()[0]\n        injection_blocked = injection_count == 0\n        print(f\"  injection attempt blocked: {injection_blocked} (returned {injection_count} rows)\")\n\n        # Cleanup\n        cur.execute(f\"DROP TABLE IF EXISTS {TABLE}\")\n        conn.commit()\n\nfinally:\n    conn.close()\n\n# ─────────────────────────────────────────\n# POST_EXECUTION\n# ─────────────────────────────────────────\n\nassert rows_returned == 2, f\"FAIL: expected 2 rows (alice=90, carol=88), got {rows_returned}\"\nassert rows[0] == (\"alice\", 90), f\"FAIL: expected ('alice', 90), got {rows[0]}\"\nassert rows[1] == (\"carol\", 88), f\"FAIL: expected ('carol', 88), got {rows[1]}\"\nassert injection_blocked, \"FAIL: SQL injection was not blocked\"\n\nresult = {\n    \"rows_returned\":    rows_returned,\n    \"injection_blocked\": injection_blocked,\n    \"parameterized_ok\": rows_returned == 2,\n}\nprint(json.dumps(result, indent=2))\nprint(\"PASS\")\n"}