ARTEX型AI攻撃にWebアプリはどう備えるべきか？Next.js・API・Supabaseの認可設計を、実際に動かして検証する

 Browser (untrusted)
    |  request + session token
    v
 [Gateway / Auth]   verifies signature, expiry, audience -> verified claims (sub)
    |
    v
 [App server]       L2: authorize(principal, action, resource)   <- TypeScript
    |   query
    v
 [PostgreSQL]       L3: Row Level Security (policies + WITH CHECK) <- SQL

 Paths AROUND L3: service role / BYPASSRLS / SECURITY DEFINER / owner

// invoice-access.ts — 前提: Node 24（型の除去で .ts を直接実行）、TypeScript strict。外部依存なし。
export type Role = "admin" | "member";
export type Action = "read" | "update";
export type Principal = { readonly userId: string; readonly roles: ReadonlyMap<string, Role> }; // tenantId -> role
export type Invoice = { readonly id: string; readonly tenantId: string; readonly amountJpy: number };

const UUID = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i;

/** 信頼境界：unknown のクレームを Principal に絞る。不正なら null（=未認証）。例外は投げない。 */
export function toPrincipal(claims: unknown, roles: ReadonlyMap<string, Role>): Principal | null {
  if (typeof claims !== "object" || claims === null) return null;
  const sub = (claims as { sub?: unknown }).sub;
  return typeof sub === "string" && UUID.test(sub) ? { userId: sub, roles } : null;
}

/** detail は監査ログ専用。HTTPレスポンスには絶対に出さない。 */
export type AuditDetail = "ok" | "unauthenticated" | "missing" | "cross_tenant" | "role_insufficient";
export type Outcome = { readonly status: 200 | 401 | 403 | 404; readonly body: { readonly error?: string; readonly invoice?: Invoice } };
export type AuditEvent = { readonly principalId: string | null; readonly action: Action; readonly invoiceId: string; readonly decision: "allow" | "deny"; readonly detail: AuditDetail };

const NOT_FOUND: Outcome = { status: 404, body: { error: "not_found" } };

export function authorize(principal: Principal | null, action: Action, invoiceId: string, store: ReadonlyMap<string, Invoice>): { outcome: Outcome; audit: AuditEvent } {
  const ev = (decision: "allow" | "deny", detail: AuditDetail): AuditEvent => ({ principalId: principal?.userId ?? null, action, invoiceId, decision, detail });
  if (!principal) return { outcome: { status: 401, body: { error: "unauthenticated" } }, audit: ev("deny", "unauthenticated") };
  const invoice = store.get(invoiceId);
  if (!invoice) return { outcome: NOT_FOUND, audit: ev("deny", "missing") };
  const role = principal.roles.get(invoice.tenantId);
  // 他テナントは「そんな請求書は無い」と外から全く同じに見える（ステータスも本文も同じ）…
  if (!role) return { outcome: NOT_FOUND, audit: ev("deny", "cross_tenant") };           // …が、監査ログでは区別する。
  if (action === "update" && role !== "admin") return { outcome: { status: 403, body: { error: "forbidden" } }, audit: ev("deny", "role_insufficient") };
  return { outcome: { status: 200, body: { invoice } }, audit: ev("allow", "ok") };
}

// invoice-access.test.ts — node --test invoice-access.test.ts
import test from "node:test";
import assert from "node:assert/strict";
import { authorize, toPrincipal, type Invoice, type Role } from "./invoice-access.ts";

const TA = "aaaaaaaa-0000-0000-0000-000000000001", TB = "bbbbbbbb-0000-0000-0000-000000000002";
const A1 = "11111111-0000-0000-0000-0000000000a1", B1 = "22222222-0000-0000-0000-0000000000b1", NONE = "99999999-0000-0000-0000-000000000000";
const store = new Map<string, Invoice>([
  [A1, { id: A1, tenantId: TA, amountJpy: 1000 }],
  [B1, { id: B1, tenantId: TB, amountJpy: 9000 }],
]);
const who = (sub: string, roles: [string, Role][]) => toPrincipal({ sub }, new Map(roles))!;
const alice = who("a1111111-0000-0000-0000-000000000001", [[TA, "admin"]]);
const amy = who("a2222222-0000-0000-0000-000000000002", [[TA, "member"]]);
const dual = who("d5555555-0000-0000-0000-000000000005", [[TA, "admin"], [TB, "member"]]);

test("unauthenticated -> 401, audit has no principal", () => {
  const r = authorize(null, "read", A1, store);
  assert.equal(r.outcome.status, 401); assert.equal(r.audit.principalId, null);
});
test("own-tenant read -> 200; admin update -> 200", () => {
  assert.equal(authorize(alice, "read", A1, store).outcome.status, 200);
  assert.equal(authorize(alice, "update", A1, store).outcome.status, 200);
});
test("cross-tenant id and nonexistent id are indistinguishable to the caller, distinguishable in the audit log", () => {
  const cross = authorize(alice, "read", B1, store), none = authorize(alice, "read", NONE, store);
  assert.deepEqual(cross.outcome, none.outcome);
  assert.equal(cross.audit.detail, "cross_tenant"); assert.equal(none.audit.detail, "missing");
});
test("cross-tenant WRITE is also 404 (never 403)", () => {
  assert.equal(authorize(alice, "update", B1, store).outcome.status, 404);
  assert.equal(authorize(amy, "update", B1, store).outcome.status, 404);
});
test("member can read but not update own tenant -> 403", () => {
  assert.equal(authorize(amy, "read", A1, store).outcome.status, 200);
  const r = authorize(amy, "update", A1, store);
  assert.equal(r.outcome.status, 403); assert.equal(r.audit.detail, "role_insufficient");
});
test("roles are per tenant: admin in A is only a member in B", () => {
  assert.equal(authorize(dual, "read", B1, store).outcome.status, 200);
  assert.equal(authorize(dual, "update", B1, store).outcome.status, 403);
  assert.equal(authorize(dual, "update", A1, store).outcome.status, 200);
});
test("denied responses never leak the invoice body", () => {
  for (const r of [authorize(alice, "read", B1, store), authorize(amy, "update", A1, store), authorize(null, "read", A1, store)])
    assert.equal(r.outcome.body.invoice, undefined);
});
test("toPrincipal fails closed on malformed claims", () => {
  for (const bad of [undefined, null, 42, "x", {}, { sub: 7 }, { sub: "not-a-uuid" }, { sub: "' or 1=1 --" }])
    assert.equal(toPrincipal(bad, new Map()), null);
});

-- Fictional B2B SaaS: two tenants, invoices. LOCAL LAB ONLY.
create role app_owner nologin;                 -- owns tables (migration role)
create role app_user  nologin;                 -- ~ Supabase "authenticated"
create role app_anon  nologin;                 -- ~ Supabase "anon"
create role app_service nologin bypassrls;     -- ~ Supabase "service_role"
create role app_definer nologin bypassrls;     -- a definer role that bypasses RLS

create schema auth;   create schema private;
create function auth.uid() returns uuid language sql stable
  as $$ select nullif(current_setting('request.jwt.claim.sub', true), '')::uuid $$;  -- 検証済みクレームを前提。SQLに直接つながるクライアントは set_config で偽装できるため、ラボ専用
grant usage on schema auth to app_owner, app_definer, app_user, app_anon, app_service;
grant execute on function auth.uid() to app_owner, app_definer, app_user, app_anon, app_service;

alter schema private owner to app_owner;
grant usage on schema public to app_owner, app_definer, app_user, app_anon, app_service;
grant create on schema public to app_owner;

set role app_owner;
create table public.tenants     (id uuid primary key, name text not null);
create table public.memberships (tenant_id uuid references public.tenants, user_id uuid not null,
                                 role text not null check (role in ('admin','member')),
                                 primary key (tenant_id, user_id));
create table public.invoices    (id uuid primary key, tenant_id uuid not null references public.tenants,
                                 customer text not null, amount_jpy integer not null check (amount_jpy >= 0),
                                 status text not null default 'draft' check (status in ('draft','sent','void')));

-- SECURITY DEFINER helper: 持ち主の権限で動く。小さく、search_path を固定し、ユーザーIDを引数に取らない。
create function private.has_role(p_tenant uuid, p_roles text[]) returns boolean
  language sql stable security definer set search_path = ''
  as $$ select exists (select 1 from public.memberships m
                        where m.tenant_id = p_tenant and m.user_id = auth.uid() and m.role = any(p_roles)) $$;
revoke all on function private.has_role(uuid, text[]) from public;
grant usage on schema private to app_user, app_definer;
grant execute on function private.has_role(uuid, text[]) to app_user, app_definer;

alter table public.tenants     enable row level security;  alter table public.tenants  force row level security;
alter table public.memberships enable row level security;   -- FORCEしない（理由は落とし穴2）
alter table public.invoices    enable row level security;  alter table public.invoices force row level security;

create policy memberships_self on public.memberships for select to app_user using (user_id = auth.uid());
create policy tenants_member   on public.tenants     for select to app_user using (private.has_role(id, array['admin','member']));
create policy inv_select on public.invoices for select to app_user
  using (private.has_role(tenant_id, array['admin','member']));
create policy inv_insert on public.invoices for insert to app_user
  with check (private.has_role(tenant_id, array['admin']));
create policy inv_update on public.invoices for update to app_user
  using (private.has_role(tenant_id, array['admin']))
  with check (private.has_role(tenant_id, array['admin']));   -- 更新後の行への要求を明示
-- DELETE のポリシーなし

grant select on public.tenants, public.memberships to app_user;
grant select, insert, update on public.invoices to app_user;     -- app_anon には何も与えない
grant select, insert, update, delete on all tables in schema public to app_service;

-- THE PITFALL: 呼び出し側のテナントを確認しない SECURITY DEFINER 関数
create function public.get_invoice_unsafe(p_id uuid) returns setof public.invoices
  language sql stable security definer set search_path = ''
  as $$ select * from public.invoices where id = p_id $$;
-- FIX: 同じ機能に、呼び出し側の所属の再確認を足す
create function public.get_invoice(p_id uuid) returns setof public.invoices
  language sql stable security definer set search_path = ''
  as $$ select i.* from public.invoices i where i.id = p_id and private.has_role(i.tenant_id, array['admin','member']) $$;
reset role;

-- (superuser) 2つの関数をRLSを回避する定義者に渡し、EXECUTEを絞る
alter function public.get_invoice_unsafe(uuid) owner to app_definer;
alter function public.get_invoice(uuid) owner to app_definer;
grant select on public.invoices to app_definer;
revoke all on function public.get_invoice_unsafe(uuid), public.get_invoice(uuid) from public;   -- 素のPostgresではこれで足りる。Supabaseではanonにも明示付与されるため from public, anon まで必要
grant execute on function public.get_invoice_unsafe(uuid), public.get_invoice(uuid) to app_user;

-- 種データ（RLSを回避するスーパーユーザーで投入）
insert into public.tenants values ('aaaaaaaa-0000-0000-0000-000000000001','Tenant A'),('bbbbbbbb-0000-0000-0000-000000000002','Tenant B');
insert into public.memberships values
 ('aaaaaaaa-0000-0000-0000-000000000001','a1111111-0000-0000-0000-000000000001','admin'),   -- alice
 ('aaaaaaaa-0000-0000-0000-000000000001','a2222222-0000-0000-0000-000000000002','member'),  -- amy
 ('bbbbbbbb-0000-0000-0000-000000000002','b3333333-0000-0000-0000-000000000003','admin');   -- bob
insert into public.invoices(id,tenant_id,customer,amount_jpy) values
 ('11111111-0000-0000-0000-0000000000a1','aaaaaaaa-0000-0000-0000-000000000001','A-Customer-1',1000),
 ('11111111-0000-0000-0000-0000000000a2','aaaaaaaa-0000-0000-0000-000000000001','A-Customer-2',2000),
 ('22222222-0000-0000-0000-0000000000b1','bbbbbbbb-0000-0000-0000-000000000002','B-Customer-1',9000);

# 1項目 = 1トランザクション。ROLLBACKするので、項目の順序に依存しない。
printf "begin;\n%s\nset local role %s;\nselect set_config('request.jwt.claim.sub','%s',true) \\\\gset\n%s\nrollback;\n" \
  "$pre" "$role" "$sub" "$sql" | $PSQL 2>&1 | grep -v -E '^(BEGIN|SET|ROLLBACK|GRANT|REVOKE|ALTER [A-Z ]+)$'
# 先頭で「aliceが自社の2件を見る」陽性対照を取り、失敗したら全体を中止する
# （セットアップが壊れていると、すべての『拒否』が偽の成功になるため）

ERROR: duplicate key value violates unique constraint "invoices_pkey"

alter table public.invoices alter column id set default gen_random_uuid();
revoke insert on public.invoices from app_user;
grant insert (tenant_id, customer, amount_jpy) on public.invoices to app_user;   -- id を含めない
revoke update on public.invoices from app_user;
grant update (customer, amount_jpy, status) on public.invoices to app_user;      -- id と tenant_id を含めない

alter table public.invoices alter column id set default gen_random_uuid();
revoke insert, update on public.invoices from anon, authenticated;
grant insert (tenant_id, customer, amount_jpy) on public.invoices to authenticated;
grant update (customer, amount_jpy, status) on public.invoices to authenticated;

select principal_id,
       count(distinct invoice_id) as distinct_denied_ids,
       min(at) as first_seen, max(at) as last_seen
from audit_events
where at >= now() - interval '5 minutes'
  and detail in ('cross_tenant', 'missing')
group by principal_id
having count(distinct invoice_id) >= 20;     -- 20は例示。自社のベースラインから決める