Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions CONTRIBUTING.md
Original file line number Diff line number Diff line change
Expand Up @@ -68,6 +68,8 @@ Rules for test-only code (see README, "Test-only switches"):
is off without `CLEANSTACK_TEST_MODE=1` and when `VERCEL=1`.
- Server code reads the user through `@/lib/auth`; eslint rejects `@clerk/nextjs/server`
imports elsewhere, so the auth bypass keeps a single entry point.
- Every AI call is preceded by `checkAiBudget(teamId, calls)` and followed by an awaited
`meterBedrockCall(...)` (`src/lib/bedrock-meter.ts`), so the guest and team caps see it.
- 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
Expand Down
19 changes: 17 additions & 2 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -238,6 +238,7 @@ aws lambda update-function-code --function-name cleanstack-executor \
| Startup log `invalid or missing environment` | Set the listed variables in Vercel and redeploy. |
| Rotating `WEBHOOK_SECRET` | Update the Vercel env **and** the profiler and ai-trigger Lambdas together. Requests fail with 401 while they differ. |
| AI spend | `GET /api/admin/ai-spend` with `x-admin-secret` lists the current month's estimated spend per team. |
| Guests see "Guest AI capacity is used up for today" | All guests together reached $5 of estimated Bedrock spend since 00:00 UTC. It resets at midnight UTC; the number is `GUEST_LIMITS.aiSpendAllGuestsPerDayUsd`. |

---

Expand All @@ -253,8 +254,22 @@ What the code does today. Settings that live only in AWS are listed as such.
- 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 `<user_data>` 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.
- **AI.** Uploaded content is sent to Amazon Bedrock and wrapped in `<user_data>` 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). `checkAiBudget` (`src/lib/bedrock-meter.ts`) runs in Postgres before every Bedrock call (suggest-transforms on every pass, the auto-validate committee, chat-builder, generate-data), and every call is metered in `bedrock_usage`.
- **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>.<expiry>.<HMAC-SHA256>`, 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`.
- **Guest limits** (all counted in Postgres, so they hold when Upstash is down):

| Limit | Value | Where |
|---|---|---|
| File size | 2 MB | presigned POST policy, `/api/upload` 413, profiler |
| Rows per run | 5,000 | `suggest-transforms`, before templates or AI; the run fails |
| Rows per guest | 10,000 (all passes; runs refused for size do not count) | `checkQuota` "guest" plan |
| Uploads | 3 per guest, 10 per IP hash per 24 h | the capped `INSERT` in `src/lib/guest-quota.ts` (also refuses expired sessions) |
| Pipelines | 5 per guest | `POST /api/pipelines` |
| AI calls | 10 per rolling hour | `checkAiBudget` |
| AI spend | $0.25 per guest; $5 per UTC day for all guests together | `checkAiBudget` |
| Sessions | 5 per IP hash and 200 overall per 24 h | `POST /api/guest` |

Guests cannot use the chat builder or synthetic data, auto-clean, training export, Slack alerts, account deletion, templates or `/api/admin`: the middleware answers 403 (or redirects `/templates` to the dashboard) and each route also refuses guest ids (`forbidGuest`). When auto-validate is over budget, the run falls back to manual review.
- **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.
Expand All @@ -271,7 +286,7 @@ Report vulnerabilities as described in [SECURITY.md](SECURITY.md).
- **No retention enforcement.** Raw files of failed runs stay in S3; `data_retention_days` is not enforced.
- **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.
- **AI budget is checked before a call and recorded after it.** Concurrent requests can each pass the check and overshoot a cap by the cost of the calls in flight (the committee's 3 calls are reserved together). Costs are estimates from `src/lib/ai-config.ts`, not the AWS bill.
- **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.
Expand Down
3 changes: 3 additions & 0 deletions src/app/api/account/route.ts
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
import { auth } from "@/lib/auth";
import { forbidGuest } from "@/lib/guest-guard";
import { NextRequest, NextResponse } from "next/server";
import { S3Client } from "@aws-sdk/client-s3";
import { query, queryOne, withTransaction } from "@/lib/db";
Expand Down Expand Up @@ -29,6 +30,8 @@ const SAFE_ID = /^[A-Za-z0-9_-]+$/;
export async function DELETE(req: NextRequest) {
const { userId } = await auth();
if (!userId) return NextResponse.json({ error: "Unauthorized" }, { status: 401 });
const guestRes = forbidGuest(userId);
if (guestRes) return guestRes;

// Require explicit confirmation to prevent accidental or CSRF-triggered deletion
const confirm = req.nextUrl.searchParams.get("confirm");
Expand Down
5 changes: 5 additions & 0 deletions src/app/api/alerts/configure/route.ts
Original file line number Diff line number Diff line change
@@ -1,10 +1,13 @@
import { NextRequest, NextResponse } from "next/server";
import { auth } from "@/lib/auth";
import { forbidGuest } from "@/lib/guest-guard";
import { queryOne, queryOneWithTeam } from "@/lib/db";

export async function POST(req: NextRequest) {
const { userId } = await auth();
if (!userId) return NextResponse.json({ error: "Unauthorized" }, { status: 401 });
const guestRes = forbidGuest(userId);
if (guestRes) return guestRes;

const { pipeline_id, slack_webhook_url } = await req.json();
if (!pipeline_id) {
Expand Down Expand Up @@ -50,6 +53,8 @@ export async function POST(req: NextRequest) {
export async function GET(req: NextRequest) {
const { userId } = await auth();
if (!userId) return NextResponse.json({ error: "Unauthorized" }, { status: 401 });
const guestRes = forbidGuest(userId);
if (guestRes) return guestRes;

const { searchParams } = new URL(req.url);
const pipeline_id = searchParams.get("pipeline_id");
Expand Down
22 changes: 18 additions & 4 deletions src/app/api/auto-validate/[runId]/route.ts
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,7 @@ import { languageModel } from "@/lib/ai-model";
import { BEDROCK_MODEL_ID } from "@/lib/ai-config";
import { SQSClient, SendMessageCommand } from "@aws-sdk/client-sqs";
import { query, queryOne } from "@/lib/db";
import { meterBedrockCall } from "@/lib/bedrock-meter";
import { checkAiBudget, meterBedrockCall } from "@/lib/bedrock-meter";
import type { TransformRule, DataProfile } from "@/lib/types";
import { requireEnv, optionalEnv, awsRegion } from "@/lib/env";
import { logger } from "@/lib/logger";
Expand Down Expand Up @@ -169,6 +169,20 @@ Include a vote for every rule_id listed. No extra text.`;
[runId]
);

// AI budget for the 3 committee calls (src/lib/bedrock-meter.ts). Over budget,
// the run falls back to manual review instead of failing: the rules exist.
if (pipelineRow?.team_id) {
const budget = await checkAiBudget(pipelineRow.team_id, 3);
if (!budget.ok) {
log.warn("AI budget reached; auto-validate falls back to manual review", { run_id: runId, scope: budget.scope });
await queryOne(
"UPDATE pipeline_runs SET status = 'awaiting_approval', error_message = $2, updated_at = now() WHERE id = $1",
[runId, `Auto-review skipped: ${budget.error} Review the rules manually.`]
);
return NextResponse.json({ ok: true, fallback: "manual_review", scope: budget.scope });
}
}

const [auditorResult, statResult, domainResult] = await Promise.all([
runConsultant(
"SafetyAuditor",
Expand Down Expand Up @@ -202,13 +216,13 @@ ${responseFormat}`,

// Meter all 3 Bedrock calls
if (pipelineRow?.team_id) {
[
await Promise.all([
{ r: auditorResult, type: "auto_validate_auditor" },
{ r: statResult, type: "auto_validate_stat" },
{ r: domainResult, type: "auto_validate_domain" },
].forEach(({ r, type }) =>
].map(({ r, type }) =>
meterBedrockCall({ teamId: pipelineRow.team_id, runId, callType: type, model: BEDROCK_MODEL_ID, usage: r.usage })
);
));
}

const [auditorVotes, statVotes, domainVotes] = [auditorResult.votes, statResult.votes, domainResult.votes];
Expand Down
12 changes: 11 additions & 1 deletion src/app/api/chat-builder/generate-data/route.ts
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,9 @@ import { generateText } from "ai";
import { languageModel } from "@/lib/ai-model";
import { aiLimiter, checkRateLimit } from "@/lib/rate-limit";
import { logger } from "@/lib/logger";
import { forbidGuest } from "@/lib/guest-guard";
import { checkAiBudget, meterBedrockCall } from "@/lib/bedrock-meter";
import { BEDROCK_MODEL_ID } from "@/lib/ai-config";

const log = logger.child({ route: "POST /api/chat-builder/generate-data" });

Expand Down Expand Up @@ -59,9 +62,15 @@ export async function POST(req: NextRequest) {
const { userId } = await auth();
if (!userId) return NextResponse.json({ error: "Unauthorized" }, { status: 401 });

const guestRes = forbidGuest(userId);
if (guestRes) return guestRes;

const rateLimitRes = await checkRateLimit(aiLimiter, userId);
if (rateLimitRes) return rateLimitRes;

const budget = await checkAiBudget(userId);
if (!budget.ok) return NextResponse.json({ error: budget.error }, { status: budget.status });

const body: GenerateRequest = await req.json();
const { description: rawDescription, config, format = "csv", row_count = 20 } = body;
const description = typeof rawDescription === "string" ? rawDescription.slice(0, 2000) : "";
Expand Down Expand Up @@ -114,11 +123,12 @@ Requirements:

let rows: Record<string, unknown>[];
try {
const { text } = await generateText({
const { text, usage } = await generateText({
model: languageModel(),
prompt,
maxOutputTokens: 4000,
});
await meterBedrockCall({ teamId: userId, runId: null, callType: "generate_data", model: BEDROCK_MODEL_ID, usage });

// Strip any accidental markdown fences
const cleaned = text.replace(/```json\s*/g, "").replace(/```\s*/g, "").trim();
Expand Down
54 changes: 54 additions & 0 deletions src/app/api/chat-builder/route.test.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,54 @@
import { beforeEach, describe, expect, it, vi } from "vitest";

const h = vi.hoisted(() => ({ budget: vi.fn(), meter: vi.fn(), streamText: vi.fn(), generateText: vi.fn() }));
vi.mock("@/lib/auth", () => ({ auth: async () => ({ userId: "user_1" }) }));
vi.mock("@/lib/bedrock-meter", () => ({ checkAiBudget: (...a: unknown[]) => h.budget(...a), meterBedrockCall: (...a: unknown[]) => h.meter(...a) }));
vi.mock("@/lib/ai-model", () => ({ languageModel: () => "model" }));
vi.mock("ai", async (orig) => ({
...(await orig<typeof import("ai")>()),
streamText: (...a: unknown[]) => h.streamText(...a),
generateText: (...a: unknown[]) => h.generateText(...a),
}));

import { POST as chat } from "./route";
import { POST as generate } from "./generate-data/route";

const chatReq = () => new Request("http://x/api/chat-builder", { method: "POST", body: JSON.stringify({ messages: [] }) });
const genReq = () => new Request("http://x/api/chat-builder/generate-data", {
method: "POST", body: JSON.stringify({ description: "orders", config: { name: "n", description: "d", rules: [] } }),
});
const OVER = { ok: false, status: 402, scope: "team_month", error: "AI spend cap reached ($200.00 / $200 this month). Contact support." };

beforeEach(() => {
vi.clearAllMocks();
h.budget.mockResolvedValue({ ok: true });
h.streamText.mockReturnValue({ toUIMessageStreamResponse: () => new Response("stream") });
h.generateText.mockResolvedValue({ text: '[{"a":1}]', usage: { inputTokens: 10, outputTokens: 5 } });
});

describe("chat-builder AI metering for signed-in users", () => {
it("checks the spend cap before streaming and meters the call on finish", async () => {
expect((await chat(chatReq())).status).toBe(200);
expect(h.budget).toHaveBeenCalledWith("user_1");
const { onFinish } = h.streamText.mock.calls[0][0];
await onFinish({ totalUsage: { inputTokens: 100, outputTokens: 20 } });
expect(h.meter).toHaveBeenCalledWith(expect.objectContaining({ teamId: "user_1", runId: null, callType: "chat_builder", usage: { inputTokens: 100, outputTokens: 20 } }));
});

it("refuses with the cap message and never calls the model", async () => {
h.budget.mockResolvedValue(OVER);
const res = await chat(chatReq());
expect(res.status).toBe(402);
expect((await res.json()).error).toBe(OVER.error);
expect(h.streamText).not.toHaveBeenCalled();
});

it("generate-data is capped and metered the same way", async () => {
expect((await generate(genReq() as never)).status).toBe(200);
expect(h.meter).toHaveBeenCalledWith(expect.objectContaining({ callType: "generate_data", usage: { inputTokens: 10, outputTokens: 5 } }));
vi.clearAllMocks();
h.budget.mockResolvedValue(OVER);
expect((await generate(genReq() as never)).status).toBe(402);
expect(h.generateText).not.toHaveBeenCalled();
});
});
12 changes: 12 additions & 0 deletions src/app/api/chat-builder/route.ts
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,9 @@ import { languageModel } from "@/lib/ai-model";
import { auth } from "@/lib/auth";
import { NextResponse } from "next/server";
import { chatLimiter, checkRateLimit } from "@/lib/rate-limit";
import { forbidGuest } from "@/lib/guest-guard";
import { checkAiBudget, meterBedrockCall } from "@/lib/bedrock-meter";
import { BEDROCK_MODEL_ID } from "@/lib/ai-config";

export const maxDuration = 60;

Expand Down Expand Up @@ -48,15 +51,24 @@ export async function POST(req: Request) {
const { userId } = await auth();
if (!userId) return NextResponse.json({ error: "Unauthorized" }, { status: 401 });

const guestRes = forbidGuest(userId);
if (guestRes) return guestRes;

const rateLimitRes = await checkRateLimit(chatLimiter, userId);
if (rateLimitRes) return rateLimitRes;

// Monthly AI spend cap, and meter the call when the stream finishes.
const budget = await checkAiBudget(userId);
if (!budget.ok) return NextResponse.json({ error: budget.error }, { status: budget.status });

const { messages }: { messages: UIMessage[] } = await req.json();

const result = streamText({
model: languageModel(),
system: SYSTEM,
messages: await convertToModelMessages(messages),
onFinish: ({ totalUsage }) =>
meterBedrockCall({ teamId: userId, runId: null, callType: "chat_builder", model: BEDROCK_MODEL_ID, usage: totalUsage }),
});

return result.toUIMessageStreamResponse();
Expand Down
3 changes: 3 additions & 0 deletions src/app/api/export-training/[runId]/route.ts
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
import { auth } from "@/lib/auth";
import { forbidGuest } from "@/lib/guest-guard";
import { NextRequest, NextResponse } from "next/server";
import { S3Client, GetObjectCommand } from "@aws-sdk/client-s3";
import { queryOneWithTeam } from "@/lib/db";
Expand Down Expand Up @@ -41,6 +42,8 @@ export async function GET(
) {
const { userId } = await auth();
if (!userId) return NextResponse.json({ error: "Unauthorized" }, { status: 401 });
const guestRes = forbidGuest(userId);
if (guestRes) return guestRes;

const rateLimitRes = await checkRateLimit(aiLimiter, userId);
if (rateLimitRes) return rateLimitRes;
Expand Down
44 changes: 44 additions & 0 deletions src/app/api/guest-blocked.test.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,44 @@
/**
* Route-level guest guards (defense in depth behind the middleware): every
* blocked route answers 403 for a guest id before touching the database, S3 or Bedrock.
*/
import { beforeEach, describe, expect, it, vi } from "vitest";
import { NextRequest } from "next/server";

const h = vi.hoisted(() => ({ userId: "guest_AAAAAAAAAAAAAAAAAAAAAA" as string | null, db: vi.fn() }));
vi.mock("@/lib/auth", () => ({ auth: async () => ({ userId: h.userId }), currentUserEmail: async () => null, userEmailById: async () => null }));
vi.mock("@/lib/db", () => {
const fail = (...a: unknown[]) => { h.db(...a); throw new Error("db must not be reached"); };
return { query: fail, queryOne: fail, queryWithTeam: fail, queryOneWithTeam: fail, withTransaction: fail };
});
vi.mock("@/lib/ai-model", () => ({ languageModel: () => { throw new Error("model must not be reached"); } }));

const ID = "11111111-2222-3333-4444-555555555555";
const runCtx = { params: Promise.resolve({ runId: ID }) };
const idCtx = { params: Promise.resolve({ id: ID }) };
const req = (method = "POST", body: unknown = {}) =>
new NextRequest("http://localhost/x", { method, headers: { "content-type": "application/json" }, body: method === "GET" ? undefined : JSON.stringify(body) });

const cases: Array<[string, () => Promise<Response>]> = [
["POST /api/chat-builder", async () => (await import("./chat-builder/route")).POST(req("POST", { messages: [] }))],
["POST /api/chat-builder/generate-data", async () => (await import("./chat-builder/generate-data/route")).POST(req())],
["POST /api/runs/:id/auto-clean", async () => (await import("./runs/[runId]/auto-clean/route")).POST(req(), runCtx)],
["GET /api/export-training/:id", async () => (await import("./export-training/[runId]/route")).GET(req("GET"), runCtx)],
["POST /api/alerts/configure", async () => (await import("./alerts/configure/route")).POST(req())],
["GET /api/alerts/configure", async () => (await import("./alerts/configure/route")).GET(req("GET"))],
["DELETE /api/account", async () => (await import("./account/route")).DELETE(req("DELETE"))],
["GET /api/templates", async () => (await import("./templates/route")).GET(req("GET"))],
["POST /api/templates", async () => (await import("./templates/route")).POST(req())],
["POST /api/templates/:id/use", async () => (await import("./templates/[id]/use/route")).POST(req(), idCtx)],
];

beforeEach(() => { h.db.mockClear(); });

describe("guest-blocked routes", () => {
it.each(cases)("%s answers 403 for a guest", async (_name, call) => {
const res = await call();
expect(res.status).toBe(403);
expect(await res.json()).toMatchObject({ guest: true });
expect(h.db).not.toHaveBeenCalled();
});
});
Loading
Loading