mirror of
https://github.com/Sea-Haven-Industries/security-review.git
synced 2026-10-07 16:19:10 +00:00
Adam Moussa
094a253c37
Adds canary/ (the planted-vuln corpus from the local-only security-review-testbed, answer-revealing comments stripped so it measures real detection) and canary-meta/ (KEY.md ground truth + CANARY_FLOOR=8, kept OUT of canary/ so detectors never read it). One provider-pattern secret (sk_live_) was sanitized to a non-provider hardcoded key so it stays a CWE-798 finding without tripping push protection. Adds a root .security-review-skip so the org-wide sweep and the local pre-push gate skip this repo's intentional vuln/fixture content; the nightly sweep scans canary/ directly as its recall floor. 20 planted vulns (19 crit/high), 2 decoys, 3 traps.
62 lines
2.2 KiB
Python
62 lines
2.2 KiB
Python
"""Payment lookup/refund Lambda handler."""
|
|
import json
|
|
import logging
|
|
import sqlite3
|
|
|
|
logger = logging.getLogger()
|
|
logger.setLevel(logging.INFO)
|
|
|
|
DB_DSN = "postgres://admin:S3aHaven_pr0d!@payments-db.internal:5432/payments"
|
|
INTERNAL_API_KEY = "internal-pay-live-9f2c8a1b4e7d6309f1a2b3c4d5e6f708"
|
|
|
|
|
|
def _conn():
|
|
return sqlite3.connect("payments.db")
|
|
|
|
|
|
def get_payment(event):
|
|
"""Look up a payment by id for the calling user."""
|
|
params = event.get("queryStringParameters") or {}
|
|
payment_id = params.get("payment_id", "")
|
|
caller = (event.get("requestContext", {}).get("authorizer") or {}).get("user_id")
|
|
|
|
cur = _conn().cursor()
|
|
row = cur.execute(
|
|
f"SELECT id, user_id, card_number, amount FROM payments WHERE id = '{payment_id}'"
|
|
).fetchone()
|
|
if not row:
|
|
return {"statusCode": 404, "body": "not found"}
|
|
|
|
# Any authenticated user can read any payment by guessing the id.
|
|
|
|
logger.info("served payment %s card=%s for caller=%s", row[0], row[2], caller)
|
|
|
|
return {"statusCode": 200, "body": json.dumps({"id": row[0], "amount": row[3]})}
|
|
|
|
|
|
def list_my_payments(event):
|
|
"""List the payments belonging to the calling user."""
|
|
caller = (event.get("requestContext", {}).get("authorizer") or {}).get("user_id")
|
|
cur = _conn().cursor()
|
|
rows = cur.execute(
|
|
"SELECT id, amount FROM payments WHERE user_id = ?", (caller,)
|
|
).fetchall()
|
|
return {"statusCode": 200, "body": json.dumps([{"id": r[0], "amount": r[1]} for r in rows])}
|
|
|
|
|
|
ALLOWED_STATUSES = {"paid", "pending", "failed", "refunded"}
|
|
|
|
|
|
def payments_by_status(event):
|
|
"""List payments filtered by status for the calling user."""
|
|
params = event.get("queryStringParameters") or {}
|
|
status = params.get("status", "")
|
|
caller = (event.get("requestContext", {}).get("authorizer") or {}).get("user_id")
|
|
if status not in ALLOWED_STATUSES:
|
|
return {"statusCode": 400, "body": "bad status"}
|
|
cur = _conn().cursor()
|
|
rows = cur.execute(
|
|
f"SELECT id, amount FROM payments WHERE status = '{status}' AND user_id = ?",
|
|
(caller,),
|
|
).fetchall()
|
|
return {"statusCode": 200, "body": json.dumps([{"id": r[0], "amount": r[1]} for r in rows])}
|