diff --git a/.env.example b/.env.example index cd5342b..bbcfa5a 100644 --- a/.env.example +++ b/.env.example @@ -35,6 +35,17 @@ CLERK_SECRET_KEY= # UPSTASH_REDIS_REST_URL=https://your-instance.upstash.io # UPSTASH_REDIS_REST_TOKEN= +# --- Guest access (optional: off unless GUEST_COOKIE_SECRET is set) ------- +# Generate with: openssl rand -base64 48 (>= 32 chars) +# GUEST_COOKIE_SECRET= +# Optional Cloudflare Turnstile on "Try as guest" (set both or neither) +# TURNSTILE_SECRET_KEY= +# NEXT_PUBLIC_TURNSTILE_SITE_KEY= + +# Per-file upload limit for signed-in users, MB (default 100; guests are fixed at 2). +# Keep in sync with MAX_UPLOAD_MB on the profiler Lambda. +# MAX_UPLOAD_MB=100 + # Structured JSON logs: debug | info | warn | error (default info) # LOG_LEVEL=info diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 87c7bb8..35a78b9 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -70,12 +70,21 @@ Rules for test-only code (see README, "Test-only switches"): imports elsewhere, so the auth bypass keeps a single entry point. - AI calls take their model from `languageModel()` (`src/lib/ai-model.ts`), never `bedrock()` directly, so CI never reaches Bedrock. +- Playwright's `webServer` inherits your shell's environment, and shell variables win over + `tests/support/test-env.mjs`. Run e2e from a clean shell: an exported `CLEANSTACK_TEST_MODE` + turns the guard servers (:3101, :3102) into test-mode servers and their tests fail. ## Conventions - **Configuration:** read environment variables through `src/lib/env.ts` (`requireEnv`, `optionalEnv`, `awsRegion`). Add new variables to `ENV_SPEC` and `.env.example`; a test enforces the latter. +- **Guests:** a `guest_…` id (`src/lib/guest.ts`) is a normal `team_id`, so tenant queries need + no change. Anything new that costs money (AI, storage, email) or is irreversible must check + `isGuestId()` and apply the limits in `src/lib/guest-limits.ts`; add a test with a guest id. +- **Uploads:** the browser uploads with a presigned POST from `/api/upload`. Size limits live in + `src/lib/upload-limits.ts` and are mirrored in `lambdas/profiler/handler.py`; change both + together. - **Logging (server):** use `logger.child({ route })` from `src/lib/logger.ts` and pass data as fields (`{ run_id, err }`). ESLint rejects `console.*` in `src/app/api` and `src/lib`. - **Secrets:** compare shared secrets with `safeCompare` from `src/lib/secrets.ts`. diff --git a/README.md b/README.md index 22602cb..76fa8a3 100644 --- a/README.md +++ b/README.md @@ -21,7 +21,8 @@ CleanStack is a web app for cleaning tabular files and documents with an AI-revi ``` Browser ── Next.js 16 app on Vercel (Clerk auth, API routes) ── Aurora PostgreSQL (RDS Data API) │ │ - │ presigned PUT │ presigned GET (120 s) for downloads + │ presigned POST │ presigned GET (120 s) for downloads + │ (size-capped) │ ▼ ▼ S3 raw bucket S3 processed bucket (output. + audit.csv per run) │ S3 event @@ -45,7 +46,7 @@ Lambda: executor ── claims run, applies rules, writes deliverable + audit fi | Component | Where it lives | Notes | |---|---|---| | Web app + API | `src/` (Next.js App Router) | Deployed on Vercel | -| Profiler Lambda | `lambdas/profiler/handler.py` | S3 PUT trigger | +| Profiler Lambda | `lambdas/profiler/handler.py` | S3 `ObjectCreated` trigger (must include `Post`; uploads are presigned POSTs) | | AI trigger Lambda | `lambdas/ai-trigger/handler.py` | SQS trigger, calls `/api/suggest-transforms` | | Executor Lambda | `lambdas/executor/handler.py` | SQS trigger | | Drift Lambda | `lambdas/drift/handler.py` | SNS trigger | @@ -128,9 +129,9 @@ PYTHON=python npm run test:e2e # tests/e2e (playwright.config. | Suite | Command | What it covers | |---|---|---| -| Integration (TS) | `npm run test:integration` | upload presigned PUT → S3 notification → profiler → suggest-transforms (fake model) → approve → executor SQS message → executor → presigned download fetch; account deletion purging every object version and delete marker; approve-rules claim under 8 concurrent requests; pg driver result shapes | +| Integration (TS) | `npm run test:integration` | upload presigned POST → S3 notification → profiler → suggest-transforms (fake model) → approve → executor SQS message → executor → presigned download fetch; account deletion purging every object version and delete marker; approve-rules claim under 8 concurrent requests; pg driver result shapes; guest sessions (signed cookie, `guest_sessions` row, per-IP cap) and a guest's 2 MB upload refused by the profiler | | Integration (Python) | `CLEANSTACK_INTEGRATION=1 pytest lambdas/tests/integration` | executor idempotency with duplicate SQS deliveries (separate and same batch), retry and last-attempt semantics, profiler on a real S3 notification | -| E2E | `npm run test:e2e` | sign in → upload a CSV → suggested rules → approve → completion → download with no `__orig_*` columns; the auth bypass is off without the flag and on Vercel | +| E2E | `npm run test:e2e` | sign in → upload a CSV → suggested rules → approve → completion → download with no `__orig_*` columns; guest session → upload → review, a guest's 2.1 MB file refused in the form and with 413; the auth bypass is off without the flag and on Vercel, and a forged guest cookie is not a session | All environment for these suites lives in `tests/support/test-env.mjs`. Variables already set in your shell win. `pytest lambdas/tests` without `CLEANSTACK_INTEGRATION=1` skips the integration directory, so the unit suite never needs containers. With the variable set, missing services are an error, not a skip. @@ -176,6 +177,9 @@ The full contract is in `src/lib/env.ts`, and `.env.example` has a placeholder f | `UPSTASH_REDIS_REST_URL`, `UPSTASH_REDIS_REST_TOKEN` | no | Rate limiting and quota cache. **Both are off when unset (fail open)** | | `SENTRY_DSN`, `NEXT_PUBLIC_SENTRY_DSN` | no | Error reporting | | `LOG_LEVEL` | no (default `info`) | Threshold of the JSON logger | +| `MAX_UPLOAD_MB` | no (default `100`) | Per-file upload limit for signed-in users. Guests are always 2 MB. Set the same value on the profiler Lambda | +| `GUEST_COOKIE_SECRET` | no | HMAC key (≥ 32 chars) for the signed `cs_guest` cookie and for hashing guest IPs. **Guest access is off while it is unset.** Generate with `openssl rand -base64 48` | +| `TURNSTILE_SECRET_KEY`, `NEXT_PUBLIC_TURNSTILE_SITE_KEY` | no | Cloudflare Turnstile check on `POST /api/guest`. Off when unset; set both or neither | Generate shared secrets with `openssl rand -hex 32`. Values shorter than 32 characters produce a startup warning. @@ -185,7 +189,7 @@ The Bedrock model id and the token prices used for metering are in `src/lib/ai-c | Function | Variables read by the code | |---|---| -| profiler | `DATABASE_URL`, `APP_URL`, `WEBHOOK_SECRET`, `SENTRY_DSN` (optional), `AWS_REGION` | +| profiler | `DATABASE_URL`, `APP_URL`, `WEBHOOK_SECRET`, `MAX_UPLOAD_MB` (optional, default 100; keep equal to the web app's), `SENTRY_DSN` (optional), `AWS_REGION` | | ai-trigger | `APP_URL`, `WEBHOOK_SECRET` | | executor | `DB_SECRET_ARN`, `S3_RAW_BUCKET`, `S3_PROCESSED_BUCKET`, `SNS_DRIFT_TOPIC_ARN` (optional), `EXECUTOR_MAX_ATTEMPTS` (optional, default 3), `SEMANTIC_DEDUP_MAX_ROWS` (optional, default 500000), `EXECUTOR_RESERVE_S` (optional, default 120: seconds of the invocation `semantic_deduplicate` leaves for writing the output, at most 25% of it), `SENTRY_DSN` (optional), `AWS_REGION` | | drift | `DB_SECRET_ARN`, `AWS_REGION` | @@ -198,7 +202,7 @@ Python dependencies are pinned in `lambdas/*/requirements.txt`. At runtime panda **Web app.** Vercel builds `main` (`npm run build`). The build needs no secrets because configuration is validated per request. Set the variables above in the Vercel project. -**Migrations.** Run `run-migration.mjs` (see above) **before** deploying code that needs new tables. Migration `002_ai_usage_tables.sql` creates `bedrock_usage` and `ai_spend_limits`. Without them, AI metering and the spend cap do not work. +**Migrations.** Run `run-migration.mjs` (see above) **before** deploying code that needs new tables. Migration `002_ai_usage_tables.sql` creates `bedrock_usage` and `ai_spend_limits`. Without them, AI metering and the spend cap do not work. Migration `003_guest_sessions.sql` creates `guest_sessions`; `POST /api/guest` returns 500 without it. **Lambdas.** Each Lambda is deployed as its handler plus its `requirements.txt`. The profiler and executor also attach the AWSSDKPandas layer. There is no build or deploy script in the repo yet, so build the zip from the pinned requirements (excluding pandas/numpy, which the layer provides) and update the function: @@ -210,6 +214,8 @@ aws lambda update-function-code --function-name cleanstack-executor \ **AWS settings the code relies on** (configure them in AWS; they cannot be checked from this repo): - **Executor SQS queue:** visibility timeout ≥ the executor Lambda timeout (AWS recommends 6×), and a dead-letter queue with `maxReceiveCount` ≥ `EXECUTOR_MAX_ATTEMPTS`. The handler does not return partial batch failures, so one failing record retries the whole batch. That is safe because execution is idempotent, but `BatchSize: 1` keeps retries simple. +- **Raw bucket CORS:** allow `POST` (and `PUT` for older clients) from the app origin. Uploads are presigned POSTs (multipart form) so that the policy's `content-length-range` makes S3 reject an oversized body. +- **Profiler S3 trigger:** the raw bucket's event notification must include `s3:ObjectCreated:Post` (or `s3:ObjectCreated:*`). A trigger on `s3:ObjectCreated:Put` alone never fires for browser uploads, and runs sit in `pending`. - **Processed bucket CORS:** allow `GET` from the app origin. The browser fetches the presigned URL for **Export As**; the plain download is a navigation and needs no CORS. - **IAM for the web app role:** `s3:PutObject` (raw), `s3:GetObject` (processed), and for account erasure `s3:ListBucketVersions`, `s3:ListBucket`, `s3:DeleteObject` and `s3:DeleteObjectVersion` on both buckets, plus `sqs:SendMessage` on both queues, `rds-data:ExecuteStatement`/`BeginTransaction`/`CommitTransaction`/`RollbackTransaction` on the cluster, `secretsmanager:GetSecretValue` on the DB secret, and `bedrock:InvokeModel` for the configured model. - **IAM for the executor:** the same S3 version permissions on the raw bucket. Without `s3:ListBucketVersions` it falls back to deleting only the two known raw keys. @@ -242,12 +248,13 @@ What the code does today. Settings that live only in AWS are listed as such. - **Authentication.** Clerk. `src/middleware.ts` protects the app pages and user API routes, and each route also calls `auth()` and returns 401 without a user. - **Tenant isolation is enforced in application code**, not by the database. Every user query filters on `pipelines.team_id = `. `queryWithTeam()` sets `app.team_id` for the transaction, but **no Row-Level Security policies are defined** in `schema.sql`, so that setting currently has no effect. A missing `team_id` predicate in a new query would leak data across tenants. - **Service-to-service calls** (Lambdas → webhooks, cron, admin) use shared secrets compared with `safeCompare` (`src/lib/secrets.ts`: SHA-256 then `timingSafeEqual`). It fails closed when the expected secret is not configured. -- **File access.** Uploads go straight to S3 with a 300 s presigned PUT; downloads use a 120 s presigned GET for the caller's own completed run. Files never pass through the serverless function. +- **File access.** Uploads go straight to S3 with a 300 s presigned POST whose policy pins the key and content type and caps the size (`content-length-range`, `src/lib/upload-limits.ts`: 100 MB or `MAX_UPLOAD_MB`, guests 2 MB). `/api/upload` refuses a larger declared size with 413 and the profiler marks a run failed, without reading the object, when the stored object is over the limit; downloads use a 120 s presigned GET for the caller's own completed run. Files never pass through the serverless function. - **Data lifecycle.** - After a successful execution, the executor deletes every object version under the run's raw prefix (`{user}/{pipeline}/{run}/`), including `extracted_text.txt` (`auto_delete_raw`, on by default). - Raw files of **failed** runs are not deleted, and there are no S3 lifecycle rules in the repo. - `DELETE /api/account?confirm=true` purges every version and delete marker under the user's raw prefix and their pipelines' processed prefixes, then deletes the DB rows. If S3 fails it aborts before touching the DB. - **AI.** Uploaded content is sent to Amazon Bedrock and wrapped in `` tags with an instruction to treat it as data. This reduces prompt-injection risk but does not prevent it, which is why rules need approval (or the committee in auto mode). The monthly AI spend cap is checked in `suggest-transforms` only. +- **Guest access** (`src/lib/guest.ts`, off unless `GUEST_COOKIE_SECRET` is set). `POST /api/guest` issues an httpOnly, `SameSite=Lax`, `Secure` cookie `cs_guest` = `guest_<22 chars>..`, valid for 24 h; `auth()` returns that `guest_…` id as the user id (it is the `team_id`), and only when no Clerk user is signed in. Sessions are recorded in `guest_sessions` with an HMAC of the client IP (never the raw IP); at most 5 per IP and 200 overall per 24 h. Turnstile is checked when `TURNSTILE_SECRET_KEY` is set. Guest quotas and blocked features are listed in `src/lib/guest-limits.ts`. - **Abuse limits.** Upstash sliding-window limits per user: 20 uploads/h, 50 AI calls/h, 30 chat messages/h. They fail open when Redis is not configured. - **Headers.** A CSP is set in `next.config.ts`. It allows `'unsafe-inline'` and `'unsafe-eval'` for scripts (needed by Next.js and Clerk without nonces). - **Secrets.** Nothing secret is committed. `.env.example` has placeholders only, and `run-migration.mjs` reads the database host from the environment. AWS account ids and ARNs from older commits remain in git history. @@ -265,7 +272,7 @@ Report vulnerabilities as described in [SECURITY.md](SECURITY.md). - **Downloads are served as stored.** Deliverables produced before the sidecar split may still contain `__orig_*` columns. **Export As** and the training export strip them; the native download does not. Re-run those pipelines to regenerate them. - **Excel blanks are not counted as nulls** by the profiler (cells are read as empty strings). Before/after scores are still computed the same way. - **AI spend cap coverage.** The cap is checked before `suggest-transforms` only. Committee calls are metered but not blocked, and chat-builder calls are neither metered nor blocked. -- **No upload size limit.** Large files can still exhaust the executor Lambda's memory or time. `semantic_deduplicate` (MinHash + LSH, a few seconds for 50k short texts) is skipped with a reason instead of running past the Lambda deadline, and above `SEMANTIC_DEDUP_MAX_ROWS` rows; other rules have no such guard. +- **Upload size limit is per file only.** 100 MB (`MAX_UPLOAD_MB`) for users, 2 MB for guests. A file under the limit can still exhaust the executor Lambda's memory or time for expensive rules. `semantic_deduplicate` (MinHash + LSH, a few seconds for 50k short texts) is skipped with a reason instead of running past the Lambda deadline, and above `SEMANTIC_DEDUP_MAX_ROWS` rows; other rules have no such guard. - **Schema drift alerts** store their diff inside `column_definitions`, so the next alert reports a phantom `_diff` column. - **Templates** copy approved rules from all recent runs of a pipeline, not only the latest run. - **`middleware.ts`** uses the file convention that Next.js 16 deprecated in favour of `proxy.ts`. It still works; the rename is pending a test against real Clerk keys. diff --git a/lambdas/profiler/handler.py b/lambdas/profiler/handler.py index e3e189b..b1436f0 100644 --- a/lambdas/profiler/handler.py +++ b/lambdas/profiler/handler.py @@ -569,6 +569,46 @@ def _effective_null_pct(s: pd.Series) -> float: return signals +# Upload size limits; keep in sync with src/lib/upload-limits.ts. +GUEST_MAX_UPLOAD_BYTES = 2 * 1024 * 1024 + + +def max_upload_bytes(key: str) -> int: + """Per-file limit for the uploader: guests (key prefix guest_) 2 MB, others MAX_UPLOAD_MB (default 100).""" + if key.startswith("guest_"): + return GUEST_MAX_UPLOAD_BYTES + try: + mb = float(os.environ.get("MAX_UPLOAD_MB", "100")) + except ValueError: + mb = 100.0 + return int((mb if mb > 0 else 100.0) * 1024 * 1024) + + +def _mb(n: int) -> str: + v = n / (1024 * 1024) + return f"{v:.0f} MB" if v == int(v) else f"{v:.1f} MB" + + +def _reject_too_large(run_id: str, size: int, limit: int) -> dict: + """Fail a pending run whose raw object exceeds the upload limit (no profiling, no AI).""" + message = f"File is {_mb(size)}; the upload limit is {_mb(limit)} per file." + print(f"[profiler] run {run_id}: {message}") + conn = get_db_conn() + cur = conn.cursor() + try: + cur.execute( + "UPDATE pipeline_runs SET status = 'failed', error_message = %s, updated_at = now() " + "WHERE id = %s AND status = 'pending' RETURNING id", + (message, run_id), + ) + claimed = cur.fetchone() + conn.commit() + finally: + cur.close() + conn.close() + return {"statusCode": 200, "run_id": run_id, "rejected": "too_large" if claimed else "already_claimed"} + + def handler(event, context): record = event["Records"][0]["s3"] bucket = record["bucket"]["name"] @@ -580,6 +620,13 @@ def handler(event, context): return {"statusCode": 200, "body": "skipped"} obj = s3.get_object(Bucket=bucket, Key=key) + size = int(obj.get("ContentLength") or 0) + limit = max_upload_bytes(key) + if size > limit: + # Backstop for the presigned POST policy (content-length-range): never read + # an oversized object into memory. Mark the run failed and stop. + obj["Body"].close() + return _reject_too_large(parts[2], size, limit) file_bytes = obj["Body"].read() # Strip UTF-8 BOM if present (prevents invisible char in first column name) if file_bytes[:3] == b'\xef\xbb\xbf': diff --git a/lambdas/tests/integration/test_profiler_s3_event.py b/lambdas/tests/integration/test_profiler_s3_event.py index 8993845..c9d3b94 100644 --- a/lambdas/tests/integration/test_profiler_s3_event.py +++ b/lambdas/tests/integration/test_profiler_s3_event.py @@ -48,3 +48,32 @@ def test_profiler_handles_real_s3_notification(live_profiler, services, db, monk finally: with db.cursor() as cur: cur.execute("DELETE FROM pipelines WHERE id = %s", (pipeline_id,)) + + +def test_profiler_rejects_an_oversized_guest_upload_without_reading_it(live_profiler, services, db, monkeypatch): + """Backstop for the presigned POST size policy (LocalStack does not enforce POST policies).""" + s3 = services["s3"] + team = "guest_" + "A" * 22 + pipeline_id, run_id = str(uuid.uuid4()), str(uuid.uuid4()) + key = f"{team}/{pipeline_id}/{run_id}/raw.csv" + with db.cursor() as cur: + cur.execute("INSERT INTO pipelines (id, name, owner_id, team_id) VALUES (%s, 'it', %s, %s)", (pipeline_id, team, team)) + cur.execute("INSERT INTO pipeline_runs (id, pipeline_id, status, file_format, raw_s3_key) VALUES (%s, %s, 'pending', 'csv', %s)", + (run_id, pipeline_id, key)) + calls = [] + monkeypatch.setattr(live_profiler.requests, "post", lambda url, **kw: calls.append(url)) + try: + body = b"id,v\n" + b"1,x\n" * (2 * 1024 * 1024 // 4 + 10) # just over 2 MB + s3.put_object(Bucket=os.environ["S3_RAW_BUCKET"], Key=key, ContentType="text/csv", Body=body) + event = {"Records": [{"s3": {"bucket": {"name": os.environ["S3_RAW_BUCKET"]}, "object": {"key": key}}}]} + assert live_profiler.handler(event, None) == {"statusCode": 200, "run_id": run_id, "rejected": "too_large"} + with db.cursor() as cur: + cur.execute("SELECT status, error_message FROM pipeline_runs WHERE id = %s", (run_id,)) + status, message = cur.fetchone() + assert status == "failed" + assert message == "File is 2.0 MB; the upload limit is 2 MB per file." + assert calls == [] # no profile-complete webhook, so no AI + finally: + with db.cursor() as cur: + cur.execute("DELETE FROM pipelines WHERE id = %s", (pipeline_id,)) + s3.delete_object(Bucket=os.environ["S3_RAW_BUCKET"], Key=key) diff --git a/lambdas/tests/test_profiler_limits.py b/lambdas/tests/test_profiler_limits.py new file mode 100644 index 0000000..59835a3 --- /dev/null +++ b/lambdas/tests/test_profiler_limits.py @@ -0,0 +1,32 @@ +"""Profiler upload size limits (mirror of src/lib/upload-limits.ts).""" +import pytest + + +def test_guest_keys_get_the_2mb_limit(profiler): + assert profiler.max_upload_bytes("guest_abcdefghijklmnopqrstuv/p/r/raw.csv") == 2 * 1024 * 1024 + + +@pytest.mark.parametrize("env,expected_mb", [(None, 100), ("50", 50), ("0.5", 0.5), ("junk", 100), ("-1", 100)]) +def test_user_limit_defaults_to_100mb_and_honours_max_upload_mb(profiler, monkeypatch, env, expected_mb): + if env is None: + monkeypatch.delenv("MAX_UPLOAD_MB", raising=False) + else: + monkeypatch.setenv("MAX_UPLOAD_MB", env) + assert profiler.max_upload_bytes("user_2abc/p/r/raw.csv") == int(expected_mb * 1024 * 1024) + + +def test_oversized_object_is_rejected_before_reading(profiler, monkeypatch): + class Body: + closed = False + def read(self): + raise AssertionError("must not read an oversized object") + def close(self): + Body.closed = True + + monkeypatch.setattr(profiler.s3, "get_object", lambda **kw: {"Body": Body(), "ContentLength": 3 * 1024 * 1024}) + rejected = [] + monkeypatch.setattr(profiler, "_reject_too_large", lambda run_id, size, limit: rejected.append((run_id, size, limit)) or {"rejected": "too_large"}) + event = {"Records": [{"s3": {"bucket": {"name": "b"}, "object": {"key": "guest_abcdefghijklmnopqrstuv/p/run-1/raw.csv"}}}]} + assert profiler.handler(event, None) == {"rejected": "too_large"} + assert rejected == [("run-1", 3 * 1024 * 1024, 2 * 1024 * 1024)] + assert Body.closed diff --git a/package-lock.json b/package-lock.json index 55ef018..2101771 100644 --- a/package-lock.json +++ b/package-lock.json @@ -16,6 +16,7 @@ "@aws-sdk/client-s3": "^3.1068.0", "@aws-sdk/client-sqs": "^3.1068.0", "@aws-sdk/rds-signer": "^3.1070.0", + "@aws-sdk/s3-presigned-post": "^3.1068.0", "@aws-sdk/s3-request-presigner": "^3.1068.0", "@clerk/nextjs": "^7.5.2", "@e965/xlsx": "^0.20.3", @@ -739,6 +740,24 @@ "node": ">=20.0.0" } }, + "node_modules/@aws-sdk/s3-presigned-post": { + "version": "3.1068.0", + "resolved": "https://registry.npmjs.org/@aws-sdk/s3-presigned-post/-/s3-presigned-post-3.1068.0.tgz", + "integrity": "sha512-c6lXKsClR1/h8jEtKgkw12O7RXWMxtQ0fasBTV2pPalkHLZfk/VSbQuKbLz3a6iXLZMq8RKhTxf/EM0mmAdA3w==", + "license": "Apache-2.0", + "dependencies": { + "@aws-sdk/client-s3": "3.1068.0", + "@aws-sdk/core": "^3.974.20", + "@aws-sdk/types": "^3.973.12", + "@smithy/core": "^3.24.6", + "@smithy/signature-v4": "^5.4.6", + "@smithy/types": "^4.14.3", + "tslib": "^2.6.2" + }, + "engines": { + "node": ">=20.0.0" + } + }, "node_modules/@aws-sdk/s3-request-presigner": { "version": "3.1068.0", "resolved": "https://registry.npmjs.org/@aws-sdk/s3-request-presigner/-/s3-request-presigner-3.1068.0.tgz", diff --git a/package.json b/package.json index 22cc41b..c5a167b 100644 --- a/package.json +++ b/package.json @@ -22,6 +22,7 @@ "@aws-sdk/client-s3": "^3.1068.0", "@aws-sdk/client-sqs": "^3.1068.0", "@aws-sdk/rds-signer": "^3.1070.0", + "@aws-sdk/s3-presigned-post": "^3.1068.0", "@aws-sdk/s3-request-presigner": "^3.1068.0", "@clerk/nextjs": "^7.5.2", "@e965/xlsx": "^0.20.3", diff --git a/src/app/api/guest/route.test.ts b/src/app/api/guest/route.test.ts new file mode 100644 index 0000000..0fd9f40 --- /dev/null +++ b/src/app/api/guest/route.test.ts @@ -0,0 +1,111 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { NextRequest } from "next/server"; + +const h = vi.hoisted(() => ({ auth: vi.fn(), queryOne: vi.fn(), turnstile: vi.fn() })); +vi.mock("@/lib/auth", () => ({ auth: () => h.auth() })); +vi.mock("@/lib/db", () => ({ queryOne: (...a: unknown[]) => h.queryOne(...a) })); +vi.mock("@/lib/turnstile", () => ({ verifyTurnstile: (...a: unknown[]) => h.turnstile(...a) })); + +import { DELETE, GET, POST } from "./route"; +import { GUEST_COOKIE, signGuestToken, verifyGuestToken } from "@/lib/guest"; +import { GUEST_LIMITS } from "@/lib/guest-limits"; + +const SECRET = "unit-test-guest-secret-0123456789abcdef"; +const req = (method: string, init: { cookie?: string; body?: unknown; ip?: string } = {}) => + new NextRequest("http://localhost/api/guest", { + method, + headers: { + "content-type": "application/json", + "x-forwarded-for": init.ip ?? "203.0.113.7", + ...(init.cookie ? { cookie: `${GUEST_COOKIE}=${init.cookie}` } : {}), + }, + body: init.body === undefined ? undefined : JSON.stringify(init.body), + }); + +beforeEach(() => { + vi.clearAllMocks(); + process.env.GUEST_COOKIE_SECRET = SECRET; + h.auth.mockResolvedValue({ userId: null }); + h.turnstile.mockResolvedValue(true); + h.queryOne.mockImplementation(async (sql: string, params: unknown[]) => + sql.startsWith("INSERT INTO guest_sessions") ? { id: params[0] } : { n: 0 }); +}); +afterEach(() => { delete process.env.GUEST_COOKIE_SECRET; }); + +describe("POST /api/guest", () => { + it("404s when guest access is off", async () => { + delete process.env.GUEST_COOKIE_SECRET; + expect((await POST(req("POST"))).status).toBe(404); + expect((await DELETE()).status).toBe(404); + expect(await (await GET(req("GET"))).json()).toEqual({ enabled: false, guest: null }); + }); + + it("issues a signed httpOnly 24 h cookie and records the session with a hashed IP and the caps", async () => { + const res = await POST(req("POST")); + expect(res.status).toBe(201); + const cookie = res.cookies.get(GUEST_COOKIE)!; + expect(cookie.httpOnly).toBe(true); + expect(cookie.sameSite).toBe("lax"); + expect(cookie.secure).toBe(true); + const session = await verifyGuestToken(cookie.value, SECRET); + expect(session).not.toBeNull(); + expect(session!.expiresAt - Math.floor(Date.now() / 1000)).toBeGreaterThan(24 * 3600 - 10); + + const [sql, params] = h.queryOne.mock.calls[0]; + expect(sql).toContain("INSERT INTO guest_sessions"); + expect(params[0]).toBe(session!.guestId); + expect(params[1]).not.toContain("203.0.113.7"); + expect(params.slice(2)).toEqual([24 * 3600, GUEST_LIMITS.guestsPerIpPerDay, GUEST_LIMITS.guestsPerDay]); + }); + + it("reuses a valid guest cookie instead of creating another session", async () => { + const token = await signGuestToken({ guestId: "guest_AAAAAAAAAAAAAAAAAAAAAA", expiresAt: Math.floor(Date.now() / 1000) + 60 }, SECRET); + const res = await POST(req("POST", { cookie: token })); + expect(res.status).toBe(200); + expect((await res.json()).reused).toBe(true); + expect(h.queryOne).not.toHaveBeenCalled(); + }); + + it("409s for a signed-in account", async () => { + h.auth.mockResolvedValue({ userId: "user_2abc" }); + expect((await POST(req("POST"))).status).toBe(409); + }); + + it("403s when Turnstile rejects the token", async () => { + h.turnstile.mockResolvedValue(false); + const res = await POST(req("POST", { body: { turnstile_token: "bad" } })); + expect(res.status).toBe(403); + expect(h.turnstile).toHaveBeenCalledWith("bad", "203.0.113.7"); + expect(h.queryOne).not.toHaveBeenCalled(); + }); + + it("429s at the per-IP cap and 503s at the global cap", async () => { + h.queryOne.mockImplementation(async (sql: string) => (sql.startsWith("INSERT") ? null : { n: GUEST_LIMITS.guestsPerIpPerDay })); + const perIp = await POST(req("POST")); + expect(perIp.status).toBe(429); + expect(perIp.cookies.get(GUEST_COOKIE)).toBeUndefined(); + + h.queryOne.mockImplementation(async (sql: string) => (sql.startsWith("INSERT") ? null : { n: 1 })); + expect((await POST(req("POST"))).status).toBe(503); + }); + + it("answers JSON 500 when the database fails", async () => { + h.queryOne.mockRejectedValue(new Error("db down")); + const res = await POST(req("POST")); + expect(res.status).toBe(500); + expect((await res.json()).error).toMatch(/Could not start a guest session/); + }); +}); + +describe("GET/DELETE /api/guest", () => { + it("reports the session and clears it", async () => { + const expiresAt = Math.floor(Date.now() / 1000) + 60; + const token = await signGuestToken({ guestId: "guest_AAAAAAAAAAAAAAAAAAAAAA", expiresAt }, SECRET); + expect(await (await GET(req("GET", { cookie: token }))).json()).toEqual({ + enabled: true, guest: { expires_at: new Date(expiresAt * 1000).toISOString() }, + }); + expect(await (await GET(req("GET", { cookie: "forged" }))).json()).toEqual({ enabled: true, guest: null }); + const del = await DELETE(); + expect(del.headers.get("set-cookie")).toMatch(new RegExp(`${GUEST_COOKIE}=;`)); + }); +}); diff --git a/src/app/api/guest/route.ts b/src/app/api/guest/route.ts new file mode 100644 index 0000000..1b7d094 --- /dev/null +++ b/src/app/api/guest/route.ts @@ -0,0 +1,104 @@ +/** + * Guest sessions (src/lib/guest.ts). Answers 404 unless GUEST_COOKIE_SECRET is set. + * + * POST { turnstile_token? } -> issues the signed `cs_guest` cookie (24 h). + * Reuses a still-valid guest cookie. 409 when signed in with an account, + * 403 when Turnstile is enabled and the token fails, 429 when this IP + * already started GUEST_LIMITS.guestsPerIpPerDay sessions in 24 h, 503 + * when GUEST_LIMITS.guestsPerDay sessions were started in 24 h overall. + * GET -> { enabled, guest: { expires_at } | null } + * DELETE -> ends the session (clears the cookie; data is purged within 24 h). + */ +import { NextRequest, NextResponse } from "next/server"; +import { auth } from "@/lib/auth"; +import { queryOne } from "@/lib/db"; +import { + GUEST_COOKIE, GUEST_TTL_SECONDS, clientIp, guestAccessEnabled, guestCookieOptions, guestFromCookie, + guestSecret, hashIp, isGuestId, newGuestId, signGuestToken, +} from "@/lib/guest"; +import { GUEST_LIMITS } from "@/lib/guest-limits"; +import { verifyTurnstile } from "@/lib/turnstile"; +import { isTestMode } from "@/lib/test-mode"; +import { logger } from "@/lib/logger"; + +const log = logger.child({ route: "/api/guest" }); + +export const dynamic = "force-dynamic"; + +const notFound = () => new NextResponse("Not Found", { status: 404 }); +const iso = (unixSeconds: number) => new Date(unixSeconds * 1000).toISOString(); + +export async function GET(req: NextRequest) { + if (!guestAccessEnabled()) return NextResponse.json({ enabled: false, guest: null }); + const session = await guestFromCookie(req.cookies.get(GUEST_COOKIE)?.value); + return NextResponse.json({ enabled: true, guest: session ? { expires_at: iso(session.expiresAt) } : null }); +} + +export async function POST(req: NextRequest) { + const secret = guestSecret(); + if (!secret) return notFound(); + + try { + const { userId } = await auth(); + if (userId && !isGuestId(userId)) { + return NextResponse.json({ error: "You are already signed in." }, { status: 409 }); + } + const existing = await guestFromCookie(req.cookies.get(GUEST_COOKIE)?.value); + if (existing) return NextResponse.json({ ok: true, expires_at: iso(existing.expiresAt), reused: true }); + + const body = (await req.json().catch(() => ({}))) as { turnstile_token?: unknown }; + const ip = clientIp(req.headers); + if (!(await verifyTurnstile(body?.turnstile_token, ip))) { + return NextResponse.json({ error: "Verification failed. Please retry." }, { status: 403 }); + } + + const ipHash = await hashIp(ip, secret); + const guestId = newGuestId(); + const expiresAt = Math.floor(Date.now() / 1000) + GUEST_TTL_SECONDS; + + // Caps checked and the row inserted in one statement. + const inserted = await queryOne<{ id: string }>( + `INSERT INTO guest_sessions (id, ip_hash, expires_at) + SELECT $1, $2, now() + make_interval(secs => $3) + WHERE (SELECT count(*) FROM guest_sessions + WHERE ip_hash = $2 AND created_at > now() - interval '24 hours') < $4 + AND (SELECT count(*) FROM guest_sessions + WHERE created_at > now() - interval '24 hours') < $5 + RETURNING id`, + [guestId, ipHash, GUEST_TTL_SECONDS, GUEST_LIMITS.guestsPerIpPerDay, GUEST_LIMITS.guestsPerDay] + ); + if (!inserted) { + const perIp = await queryOne<{ n: number }>( + "SELECT count(*)::int AS n FROM guest_sessions WHERE ip_hash = $1 AND created_at > now() - interval '24 hours'", + [ipHash] + ); + if (Number(perIp?.n ?? 0) >= GUEST_LIMITS.guestsPerIpPerDay) { + log.warn("guest session refused: per-IP cap", { cap: GUEST_LIMITS.guestsPerIpPerDay }); + return NextResponse.json( + { error: `Guest limit reached for your network (${GUEST_LIMITS.guestsPerIpPerDay} guest sessions per day). Sign up for a free account to continue.` }, + { status: 429 } + ); + } + log.warn("guest session refused: global cap", { cap: GUEST_LIMITS.guestsPerDay }); + return NextResponse.json( + { error: "Guest access is at capacity today. Please sign up for a free account or try again tomorrow." }, + { status: 503 } + ); + } + + const res = NextResponse.json({ ok: true, expires_at: iso(expiresAt) }, { status: 201 }); + res.cookies.set(GUEST_COOKIE, await signGuestToken({ guestId, expiresAt }, secret), guestCookieOptions(expiresAt, !isTestMode())); + log.info("guest session started"); + return res; + } catch (err) { + log.error("guest session failed", { err }); + return NextResponse.json({ error: "Could not start a guest session. Please retry." }, { status: 500 }); + } +} + +export async function DELETE() { + if (!guestAccessEnabled()) return notFound(); + const res = NextResponse.json({ ok: true }); + res.cookies.delete(GUEST_COOKIE); + return res; +} diff --git a/src/app/api/upload/presign.test.ts b/src/app/api/upload/presign.test.ts index ac6bcea..114a1dd 100644 --- a/src/app/api/upload/presign.test.ts +++ b/src/app/api/upload/presign.test.ts @@ -1,26 +1,37 @@ import { describe, expect, it } from "vitest"; -import { S3Client, PutObjectCommand } from "@aws-sdk/client-s3"; -import { getSignedUrl } from "@aws-sdk/s3-request-presigner"; +import { S3Client } from "@aws-sdk/client-s3"; +import { createPresignedPost } from "@aws-sdk/s3-presigned-post"; import { readFileSync } from "node:fs"; import { fileURLToPath } from "node:url"; const credentials = { accessKeyId: "AKIDEXAMPLE", secretAccessKey: "secret" }; -const presign = (client: S3Client) => - getSignedUrl(client, new PutObjectCommand({ Bucket: "b", Key: "u/p/r/raw.csv", ContentType: "text/csv" }), { expiresIn: 300 }); -describe("upload presigned PUT URL", () => { - it("the SDK default bakes an empty-body checksum into the URL (why the route opts out)", async () => { - const url = await presign(new S3Client({ region: "us-east-1", credentials })); - expect(url).toContain("x-amz-checksum-crc32=AAAAAA%3D%3D"); +describe("upload presigned POST", () => { + it("signs a policy that S3 enforces: exact key, size range and content type", async () => { + // Real AWS endpoint even when the shell points the SDK at LocalStack. + const client = new S3Client({ region: "us-east-1", credentials, requestChecksumCalculation: "WHEN_REQUIRED", endpoint: "https://s3.us-east-1.amazonaws.com" }); + const { url, fields } = await createPresignedPost(client, { + Bucket: "cleanstack-raw-test", Key: "u/p/r/raw.csv", Expires: 300, + Conditions: [["content-length-range", 1, 2 * 1024 * 1024], ["eq", "$Content-Type", "text/csv"]], + Fields: { "Content-Type": "text/csv" }, + }); + expect(url).toMatch(/^https:\/\/cleanstack-raw-test\.s3\.(us-east-1\.)?amazonaws\.com\/$/); + expect(fields.key).toBe("u/p/r/raw.csv"); + const policy = JSON.parse(Buffer.from(fields.Policy, "base64").toString("utf8")); + expect(policy.conditions).toEqual(expect.arrayContaining([ + ["content-length-range", 1, 2 * 1024 * 1024], + ["eq", "$Content-Type", "text/csv"], + { key: "u/p/r/raw.csv" }, + { bucket: "cleanstack-raw-test" }, + ])); + expect(Object.keys(fields).join(",")).not.toMatch(/checksum/i); }); - it("with WHEN_REQUIRED the URL carries no checksum, so any body can be uploaded", async () => { - const url = await presign(new S3Client({ region: "us-east-1", credentials, requestChecksumCalculation: "WHEN_REQUIRED" })); - expect(url).not.toMatch(/x-amz-checksum|x-amz-sdk-checksum-algorithm/); - }); - - it("the upload route's presigning client uses WHEN_REQUIRED", () => { + it("the upload route presigns a POST (not a PUT) with a WHEN_REQUIRED client", () => { const src = readFileSync(fileURLToPath(new URL("./route.ts", import.meta.url)), "utf8"); expect(src).toMatch(/new S3Client\(\{[^}]*requestChecksumCalculation: "WHEN_REQUIRED"/); + expect(src).toContain("createPresignedPost("); + expect(src).toContain('["content-length-range", 1, maxBytes]'); + expect(src).not.toContain("PutObjectCommand"); }); }); diff --git a/src/app/api/upload/route.test.ts b/src/app/api/upload/route.test.ts index 5d3fce4..8e7c019 100644 --- a/src/app/api/upload/route.test.ts +++ b/src/app/api/upload/route.test.ts @@ -7,7 +7,7 @@ const h = vi.hoisted(() => ({ limit: vi.fn(), queryOneWithTeam: vi.fn(), queryOne: vi.fn(), - getSignedUrl: vi.fn(), + createPresignedPost: vi.fn(), })); vi.mock("@/lib/auth", () => ({ auth: () => h.auth(), currentUserEmail: () => h.email() })); @@ -16,7 +16,7 @@ vi.mock("@/lib/db", () => ({ queryOneWithTeam: (...a: unknown[]) => h.queryOneWithTeam(...a), queryOne: (...a: unknown[]) => h.queryOne(...a), })); -vi.mock("@aws-sdk/s3-request-presigner", () => ({ getSignedUrl: (...a: unknown[]) => h.getSignedUrl(...a) })); +vi.mock("@aws-sdk/s3-presigned-post", () => ({ createPresignedPost: (...a: unknown[]) => h.createPresignedPost(...a) })); // Real checkRateLimit, with a limiter whose Upstash calls we control. vi.mock("@/lib/rate-limit", async (importOriginal) => ({ ...(await importOriginal()), @@ -27,10 +27,11 @@ const savedBucket = process.env.S3_RAW_BUCKET; process.env.S3_RAW_BUCKET = "raw-bucket"; afterAll(() => { if (savedBucket === undefined) delete process.env.S3_RAW_BUCKET; else process.env.S3_RAW_BUCKET = savedBucket; }); -const { POST } = await import("./route"); +const { GET, POST } = await import("./route"); +const POST_UPLOAD = { url: "https://raw-bucket.s3.amazonaws.com/", fields: { key: "k", Policy: "p" } }; const OK_QUOTA = { blocked: false, used: 0, includedRows: 1000, plan: "free" }; -const call = (body: unknown = { pipeline_id: "11111111-2222-3333-4444-555555555555", filename: "Email-Tracking.csv" }) => +const call = (body: unknown = { pipeline_id: "11111111-2222-3333-4444-555555555555", filename: "Email-Tracking.csv", size: 2_500 }) => POST(new Request("http://x/api/upload", { method: "POST", body: typeof body === "string" ? body : JSON.stringify(body) }) as never); beforeEach(() => { @@ -41,14 +42,53 @@ beforeEach(() => { h.limit.mockResolvedValue({ success: true, limit: 20, remaining: 19, reset: 0 }); h.queryOneWithTeam.mockResolvedValue({ id: "p" }); h.queryOne.mockResolvedValue({ id: "run_1" }); - h.getSignedUrl.mockResolvedValue("https://raw-bucket.s3.amazonaws.com/put"); + h.createPresignedPost.mockResolvedValue(POST_UPLOAD); +}); + +describe("GET /api/upload", () => { + it("returns the caller's per-file limit", async () => { + expect(await (await GET()).json()).toEqual({ max_bytes: 100 * 1024 * 1024 }); + h.auth.mockResolvedValue({ userId: "guest_" + "a".repeat(22) }); + expect(await (await GET()).json()).toEqual({ max_bytes: 2 * 1024 * 1024 }); + h.auth.mockResolvedValue({ userId: null }); + expect((await GET()).status).toBe(401); + }); }); describe("POST /api/upload", () => { - it("returns a presigned URL", async () => { + it("returns a presigned POST whose policy caps the size and pins the content type", async () => { + const res = await call(); + expect(res.status).toBe(200); + expect(await res.json()).toMatchObject({ upload: POST_UPLOAD, run_id: "run_1", max_bytes: 100 * 1024 * 1024 }); + const [, params] = h.createPresignedPost.mock.calls[0]; + expect(params).toMatchObject({ + Bucket: "raw-bucket", + Key: expect.stringMatching(/^user_1\/11111111-2222-3333-4444-555555555555\/[0-9a-f-]{36}\/raw\.csv$/), + Conditions: [["content-length-range", 1, 100 * 1024 * 1024], ["eq", "$Content-Type", "text/csv"]], + Fields: { "Content-Type": "text/csv" }, + Expires: 300, + }); + }); + + it("guests get a 2 MB policy", async () => { + h.auth.mockResolvedValue({ userId: "guest_" + "a".repeat(22) }); const res = await call(); expect(res.status).toBe(200); - expect(await res.json()).toMatchObject({ presigned_url: "https://raw-bucket.s3.amazonaws.com/put", run_id: "run_1" }); + expect(h.createPresignedPost.mock.calls[0][1].Conditions[0]).toEqual(["content-length-range", 1, 2 * 1024 * 1024]); + }); + + it("refuses a declared size over the limit with 413 before creating a run", async () => { + h.auth.mockResolvedValue({ userId: "guest_" + "a".repeat(22) }); + const res = await call({ pipeline_id: "p", filename: "big.csv", size: 2 * 1024 * 1024 + 1 }); + expect(res.status).toBe(413); + expect(await res.json()).toEqual({ error: "File is 2.0 MB; the limit is 2 MB per file.", max_bytes: 2 * 1024 * 1024 }); + expect(h.queryOne).not.toHaveBeenCalled(); + }); + + it("requires the file size", async () => { + const res = await call({ pipeline_id: "p", filename: "a.csv" }); + expect(res.status).toBe(400); + expect(h.queryOne).not.toHaveBeenCalled(); }); it("still uploads when Upstash fails DNS (the production 500 with an empty body)", async () => { @@ -82,7 +122,7 @@ describe("POST /api/upload", () => { }); it("answers JSON when presigning fails", async () => { - h.getSignedUrl.mockRejectedValue(new Error("no credentials")); + h.createPresignedPost.mockRejectedValue(new Error("no credentials")); const res = await call(); expect(res.status).toBe(500); expect((await res.json()).stage).toBe("presign"); diff --git a/src/app/api/upload/route.ts b/src/app/api/upload/route.ts index 39fa365..a5ff8e4 100644 --- a/src/app/api/upload/route.ts +++ b/src/app/api/upload/route.ts @@ -1,23 +1,27 @@ import { auth, currentUserEmail } from "@/lib/auth"; import { NextRequest, NextResponse } from "next/server"; import { randomUUID } from "crypto"; -import { S3Client, PutObjectCommand } from "@aws-sdk/client-s3"; -import { getSignedUrl } from "@aws-sdk/s3-request-presigner"; +import { S3Client } from "@aws-sdk/client-s3"; +import { createPresignedPost } from "@aws-sdk/s3-presigned-post"; import { queryOne, queryOneWithTeam } from "@/lib/db"; import { getCachedQuota } from "@/lib/quota-cache"; import { uploadLimiter, checkRateLimit } from "@/lib/rate-limit"; import type { PipelineRun } from "@/lib/types"; import { requireEnv, awsRegion } from "@/lib/env"; import { logger } from "@/lib/logger"; +import { formatMb, maxUploadBytesFor } from "@/lib/upload-limits"; const log = logger.child({ route: "POST /api/upload" }); -// Presigning only. Since SDK 3.729 the default (WHEN_SUPPORTED) bakes a CRC32 of the -// *empty* body (x-amz-checksum-crc32=AAAAAA==) into presigned PutObject URLs, so the -// browser's real upload does not match it (LocalStack rejects it with 400; see -// aws/aws-sdk-js-v3#6810). WHEN_REQUIRED restores the pre-3.729 URL shape. +// Presigning only. The upload is a presigned POST (not PUT) so the policy can carry +// content-length-range: S3 itself rejects a body over the limit. WHEN_REQUIRED keeps +// the SDK from adding checksum fields the browser's body would not match +// (aws/aws-sdk-js-v3#6810). const s3 = new S3Client({ region: awsRegion(), requestChecksumCalculation: "WHEN_REQUIRED" }); +/** Presigned POST lifetime. */ +const UPLOAD_EXPIRES_SECONDS = 300; + const ALLOWED_EXTENSIONS = new Set([ "csv", "tsv", "txt", "json", "jsonl", "xlsx", "xls", "xml", @@ -51,7 +55,11 @@ const STAGE_MESSAGES: Record = { }; /** - * Creates a pending run and returns a presigned S3 PUT URL for the raw file. + * Creates a pending run and returns a presigned S3 POST (url + form fields) for + * the raw file. Body: { pipeline_id, filename, size } (size in bytes). A file + * over the caller's limit (src/lib/upload-limits.ts: 100 MB, guests 2 MB) is + * refused here with 413; the POST policy's content-length-range makes S3 + * enforce the same limit on the actual bytes, and the profiler checks again. * * Every step runs inside one try/catch. Previously auth, rate limiting, the * Clerk email lookup and the quota query ran outside it, so any throw there @@ -62,6 +70,13 @@ const STAGE_MESSAGES: Record = { * that failed and are logged with the same `stage`, so Vercel runtime logs * show the underlying error. */ +/** The caller's per-file limit, so the upload form can refuse a large file before creating anything. */ +export async function GET() { + const { userId } = await auth(); + if (!userId) return NextResponse.json({ error: "Unauthorized" }, { status: 401 }); + return NextResponse.json({ max_bytes: maxUploadBytesFor(userId) }); +} + export async function POST(req: NextRequest) { let stage: Stage = "auth"; try { @@ -95,11 +110,21 @@ export async function POST(req: NextRequest) { stage = "parse_body"; const body = await req.json().catch(() => null); - const { pipeline_id, filename } = (body ?? {}) as { pipeline_id?: string; filename?: string }; + const { pipeline_id, filename, size } = (body ?? {}) as { pipeline_id?: string; filename?: string; size?: unknown }; if (!pipeline_id || !filename) { return NextResponse.json({ error: "pipeline_id and filename required" }, { status: 400 }); } + const maxBytes = maxUploadBytesFor(userId); + if (typeof size !== "number" || !Number.isFinite(size) || size <= 0) { + return NextResponse.json({ error: "size (file size in bytes) required" }, { status: 400 }); + } + if (size > maxBytes) { + return NextResponse.json( + { error: `File is ${formatMb(size)}; the limit is ${formatMb(maxBytes)} per file.`, max_bytes: maxBytes }, + { status: 413 } + ); + } const ext = filename.split(".").pop()?.toLowerCase() ?? ""; if (!ALLOWED_EXTENSIONS.has(ext)) { @@ -131,15 +156,18 @@ export async function POST(req: NextRequest) { if (!run) return NextResponse.json({ error: "Failed to create run" }, { status: 500 }); stage = "presign"; - const command = new PutObjectCommand({ + const upload = await createPresignedPost(s3, { Bucket: requireEnv("S3_RAW_BUCKET"), Key: s3Key, - ContentType: content_type, + Conditions: [ + ["content-length-range", 1, maxBytes], + ["eq", "$Content-Type", content_type], + ], + Fields: { "Content-Type": content_type }, + Expires: UPLOAD_EXPIRES_SECONDS, }); - const presignedUrl = await getSignedUrl(s3, command, { expiresIn: 300 }); - - return NextResponse.json({ presigned_url: presignedUrl, run_id: run.id, s3_key: s3Key }); + return NextResponse.json({ upload, max_bytes: maxBytes, run_id: run.id, s3_key: s3Key }); } catch (err) { log.error("upload failed", { stage, err }); return NextResponse.json( diff --git a/src/app/pipelines/new/page.tsx b/src/app/pipelines/new/page.tsx index 0bfcedf..743d6e5 100644 --- a/src/app/pipelines/new/page.tsx +++ b/src/app/pipelines/new/page.tsx @@ -1,10 +1,11 @@ "use client"; -import { useState, useRef } from "react"; +import { useEffect, useState, useRef } from "react"; import { useRouter } from "next/navigation"; import Nav from "@/components/Nav"; import PipelineChat from "@/components/PipelineChat"; import { readJson } from "@/lib/http"; +import { formatMb } from "@/lib/upload-limits"; import { Upload, Sparkles, @@ -30,6 +31,7 @@ const FORMAT_LABELS: Record = { }; type Tab = "upload" | "chat"; +type PresignedPost = { url: string; fields: Record }; type Step = "form" | "uploading" | "processing"; export default function NewPipelinePage() { @@ -44,6 +46,16 @@ export default function NewPipelinePage() { const [step, setStep] = useState("form"); const [error, setError] = useState(null); const [progress, setProgress] = useState(0); + // Per-file limit from GET /api/upload (100 MB by default, 2 MB for guests). The + // server and S3 enforce it regardless; this only saves a doomed upload. + const [maxBytes, setMaxBytes] = useState(null); + + useEffect(() => { + fetch("/api/upload") + .then((r) => (r.ok ? r.json() : null)) + .then((d: { max_bytes?: number } | null) => { if (d?.max_bytes) setMaxBytes(d.max_bytes); }) + .catch(() => {}); + }, []); function handleFile(f: File) { const ext = f.name.split(".").pop()?.toLowerCase() ?? ""; @@ -51,6 +63,10 @@ export default function NewPipelinePage() { setError(`Unsupported file type: .${ext}`); return; } + if (maxBytes !== null && f.size > maxBytes) { + setError(`File is ${formatMb(f.size)}; the limit is ${formatMb(maxBytes)} per file.`); + return; + } setFile(f); setError(null); } @@ -83,22 +99,25 @@ export default function NewPipelinePage() { body: JSON.stringify({ pipeline_id: pipeline.id, filename: file.name, - content_type: file.type || "application/octet-stream", + size: file.size, }), }); - const { presigned_url, run_id } = await readJson<{ presigned_url: string; run_id: string }>(uploadRes, "Upload"); + const { upload, run_id } = await readJson<{ upload: PresignedPost; run_id: string }>(uploadRes, "Upload"); setProgress(40); const uploadController = new AbortController(); const uploadTimeout = setTimeout(() => uploadController.abort(), 120_000); try { - const s3Res = await fetch(presigned_url, { - method: "PUT", - body: file, - headers: { "Content-Type": file.type || "application/octet-stream" }, - signal: uploadController.signal, - }); - if (!s3Res.ok) throw new Error(`File upload to storage failed (HTTP ${s3Res.status}) — please retry`); + // Presigned POST: policy fields first, the file last (S3 ignores fields after it). + const form = new FormData(); + for (const [k, v] of Object.entries(upload.fields)) form.append(k, v); + form.append("file", file); + const s3Res = await fetch(upload.url, { method: "POST", body: form, signal: uploadController.signal }); + if (!s3Res.ok) { + const detail = await s3Res.text().catch(() => ""); + if (/EntityTooLarge/.test(detail)) throw new Error("File is larger than your upload limit"); + throw new Error(`File upload to storage failed (HTTP ${s3Res.status}) — please retry`); + } } catch (fetchErr) { if (fetchErr instanceof DOMException && fetchErr.name === "AbortError") { throw new Error("Upload timed out — check your connection and retry"); @@ -274,7 +293,7 @@ export default function NewPipelinePage() {