Merge branch 'fix/backup-rotation-and-deletion' into release/v1.39.3

# Conflicts:
#	docs/api/openapi.yaml
#	src/lib/jobs/apple-health-import-worker.ts
This commit is contained in:
Marc-André Bombeck committed 2026-09-26 21:48:20 +02:00
commit 4409529c89
74 files changed
+4022 -307

No files matched your search

@@ -22,9 +22,15 @@ import {
storedBackupRefusal,
STORED_BACKUP_SELECT,
} from "@/lib/export/stored-backup";
import {
assessBackupKeys,
BACKUP_KEY_MISSING_CODE,
describeBackupKeyProblem,
} from "@/lib/export/backup-key-ids";
import {
readStreamedBackup,
type BackupSource,
type StreamedBackup,
} from "@/lib/export/streamed-backup";
import { annotate } from "@/lib/logging/context";
import {
@@ -72,8 +78,9 @@ export const GET = apiHandler(
let payload;
let measurementCount: number;
let streamed: StreamedBackup;
try {
const streamed = await readStreamedBackup(source);
streamed = await readStreamedBackup(source);
measurementCount = streamed.measurementCount;
payload = parseBackupPayload(streamed.raw);
} catch (err) {
@@ -93,6 +100,22 @@ export const GET = apiHandler(
);
}
// The preview is where the operator decides, so it says what the restore
// would refuse: a key the file's inner ciphertext needs and this host
// does not have. The instance settings are left out here, as the restore
// leaves them out unless asked; the restore itself checks them when they
// are asked for.
const keyVerdict = assessBackupKeys(streamed.keys, {
ignoreSections: new Set(["appSettings"]),
});
const keyProblem = describeBackupKeyProblem(keyVerdict);
if (keyProblem) {
return apiError(keyProblem, 422, {
errorCode: BACKUP_KEY_MISSING_CODE,
keyIds: [...keyVerdict.missing, ...keyVerdict.unreadable],
});
}
const summary = {
...summarizeBackup(payload),
measurements: measurementCount,
@@ -29,12 +29,23 @@ vi.mock("@/lib/db", () => ({
prisma: {
$queryRaw: (...a: unknown[]) => queryRawMock(...a),
user: { findMany: (...a: unknown[]) => findManyMock(...a) },
offhostPurgeRequest: {
count: async () => 2,
findFirst: async () => ({
requestedAt: new Date("2026-09-20T10:00:00.000Z"),
lastFailure: "AccessDenied",
}),
},
},
}));
const configuredMock = vi.fn(() => true);
vi.mock("@/lib/jobs/offhost-backup", () => ({
offhostBackupConfigured: () => configuredMock(),
probeOffhostLifecycle: async () => ({
state: "missing",
expirationDays: null,
}),
}));
import { GET } from "../route";
@@ -139,10 +150,12 @@ describe("GET /api/admin/backups — off-host freshness", () => {
const body = await read();
expect(body.offhost).toEqual({
expect(body.offhost).toMatchObject({
configured: false,
periodHours: 24,
rows: [],
// No bucket to ask on a host that is not configured.
lifecycle: { state: "unknown", expirationDays: null },
});
});
@@ -196,4 +209,20 @@ describe("GET /api/admin/backups — off-host freshness", () => {
]);
expect(JSON.stringify(body.offhost)).not.toMatch(/bucket|endpoint|secret/i);
});
it("carries the bucket's lifecycle verdict and the deletions it still owes", async () => {
configuredMock.mockReturnValue(true);
queryRawMock.mockResolvedValue([]);
findManyMock.mockResolvedValue([]);
const body = await read();
expect(body.offhost.lifecycle).toEqual({
state: "missing",
expirationDays: null,
});
expect(body.offhost.pendingDeletions).toEqual({
count: 2,
oldestRequestedAt: "2026-09-20T10:00:00.000Z",
lastFailure: "AccessDenied",
});
});
});
+21 -1
View File
@@ -27,7 +27,10 @@ import { annotate } from "@/lib/logging/context";
import { DATA_BACKUP_QUEUE } from "@/lib/jobs/data-backup-policy";
import { readLastQueueRun } from "@/lib/jobs/job-failures";
import { summariseBackupSchedule } from "@/lib/jobs/backup-schedule-status";
import { offhostBackupConfigured } from "@/lib/jobs/offhost-backup";
import {
offhostBackupConfigured,
probeOffhostLifecycle,
} from "@/lib/jobs/offhost-backup";
import {
classifyOffhostBackup,
OFFHOST_BACKUP_PERIOD_HOURS,
@@ -164,6 +167,17 @@ export const GET = apiHandler(async () => {
a.username.localeCompare(b.username),
);
const [purgeCount, oldestPurge, lifecycle] = await Promise.all([
prisma.offhostPurgeRequest.count(),
prisma.offhostPurgeRequest.findFirst({
orderBy: { requestedAt: "asc" },
select: { requestedAt: true, lastFailure: true },
}),
configured
? probeOffhostLifecycle()
: Promise.resolve({ state: "unknown" as const, expirationDays: null }),
]);
const payload: BackupsList = {
rows: list,
// Matches the retention window the backup-prune job enforces so the
@@ -182,6 +196,12 @@ export const GET = apiHandler(async () => {
// An unconfigured host has nothing to list, and the card says so rather
// than painting an empty table that reads as "no problems".
rows: configured ? offhostRows : [],
pendingDeletions: {
count: purgeCount,
oldestRequestedAt: oldestPurge?.requestedAt.toISOString() ?? null,
lastFailure: oldestPurge?.lastFailure ?? null,
},
lifecycle,
},
};
+42 -38
View File
@@ -29,7 +29,6 @@
* Phase B1 / criterion 2 of the v1.4.15 backup-completeness work.
*/
import { Readable } from "node:stream";
import { createGunzip } from "node:zlib";
import { BodyTooLargeError, readBoundedBody } from "@/lib/labs/ocr-upload";
import { NextRequest } from "next/server";
@@ -44,6 +43,16 @@ import {
import { apiError, apiSuccess, getClientIp } from "@/lib/api-response";
import { auditLog } from "@/lib/auth/audit";
import { BackupJsonError, scanBackupJson } from "@/lib/export/backup-json-scan";
import {
BackupUploadDecodeError,
decodeBackupUpload,
} from "@/lib/export/backup-upload-decode";
import {
assessBackupKeys,
BACKUP_KEY_MISSING_CODE,
BackupKeyIdCollector,
describeBackupKeyProblem,
} from "@/lib/export/backup-key-ids";
import {
BACKUP_UPLOAD_TOO_LARGE_CODE,
BackupBlobTooLargeError,
@@ -101,50 +110,22 @@ class UploadRefused extends Error {
}
}
/** Bytes → bytes, gunzipped when the file starts with the gzip magic. */
/** The file's bytes, gunzipped and counted on both ends. */
async function* decodedBytes(
source: AsyncIterable<Uint8Array>,
limit: number,
): AsyncGenerator<Uint8Array> {
let seen = 0;
async function* counted() {
for await (const chunk of source) {
seen += chunk.byteLength;
if (seen > limit) {
throw new UploadRefused(
413,
`Upload exceeds ${Math.round(limit / 1024 / 1024)} MB limit`,
{ reason: "file_size_exceeded", size: seen },
);
}
yield chunk;
}
}
const iterator = counted()[Symbol.asyncIterator]();
const first = await iterator.next();
if (first.done) return;
const rest = (async function* () {
yield first.value;
for (;;) {
const next = await iterator.next();
if (next.done) return;
yield next.value;
}
})();
const head = first.value;
if (head.byteLength >= 2 && head[0] === 0x1f && head[1] === 0x8b) {
const gunzip = Readable.from(rest).pipe(createGunzip());
try {
for await (const chunk of gunzip) yield chunk as Buffer;
} catch (err) {
if (err instanceof UploadRefused) throw err;
throw new UploadRefused(422, "Uploaded file is not valid gzip", {
reason: "invalid_gzip",
try {
yield* decodeBackupUpload(source, { compressedLimit: limit });
} catch (err) {
if (err instanceof BackupUploadDecodeError) {
throw new UploadRefused(err.status, err.message, {
reason: err.reason,
...(err.size !== undefined ? { size: err.size } : {}),
});
}
return;
throw err;
}
yield* rest;
}
export const POST = apiHandler(async (request: NextRequest) => {
@@ -259,6 +240,7 @@ export const POST = apiHandler(async (request: NextRequest) => {
}
}
let firstWithoutId: number | null = null;
const keys = new BackupKeyIdCollector();
let scanned;
try {
scanned = await scanBackupJson(storedAsRead(), {
@@ -275,6 +257,7 @@ export const POST = apiHandler(async (request: NextRequest) => {
if (!parsed.data.id && firstWithoutId === null) {
firstWithoutId = index;
}
keys.visit(element, "measurements");
},
});
} catch (err) {
@@ -343,6 +326,27 @@ export const POST = apiHandler(async (request: NextRequest) => {
);
}
// A file whose inner ciphertext needs a key this host does not have
// can be stored but never restored: the restore would refuse it, or
// worse, a restore on a host that had lost the key would write rows
// nobody can open. Refused here, where the operator is looking, with
// the key named. The instance settings are left to the restore, which
// checks them only when they are asked for back.
keys.visit(scanned.document);
const keyVerdict = assessBackupKeys(keys, {
ignoreSections: new Set(["appSettings"]),
});
const keyProblem = describeBackupKeyProblem(keyVerdict);
if (keyProblem) {
const keyIds = [...keyVerdict.missing, ...keyVerdict.unreadable];
throw new UploadRefused(
422,
keyProblem,
{ reason: "key_missing", keyIds },
{ errorCode: BACKUP_KEY_MISSING_CODE, keyIds },
);
}
// Make sure the userId on the file points at a user that exists in
// this DB. Restore would fail later anyway, but failing here gives
// the admin a precise error before any side-effect.
@@ -19,6 +19,10 @@ import { apiHandler, requireAdmin } from "@/lib/api-handler";
import { apiSuccess } from "@/lib/api-response";
import { annotate } from "@/lib/logging/context";
import { getConfiguredKeyIds } from "@/lib/crypto";
import {
offhostBackupConfigured,
probeOffhostLifecycle,
} from "@/lib/jobs/offhost-backup";
import {
scanCorpus,
type CorpusClient,
@@ -100,6 +104,100 @@ async function readRotationState(): Promise<RotationState> {
};
}
/**
* Which keys the backups still need.
*
* Rotation re-encrypts every row in the database and re-seals every stored
* backup's envelope, but the content INSIDE a backup keeps the key it was
* written under: a restore writes it back verbatim. So "every row is on the
* active key" is not the same as "the old key can go". This answers the second
* question per key id, from what each copy recorded as it was written: the
* stored copies that need it and the oldest of them, and for the off-host
* bucket the last night an object needing it went in, plus how long the
* bucket's lifecycle rule keeps objects after that.
*/
export interface BackupKeyNeeds {
stored: Array<{ keyId: string; copies: number; oldestAt: string }>;
/** Copies written before v1.39.3, which recorded no key ids. */
unrecorded: { copies: number; oldestAt: string | null };
offhost: Array<{
keyId: string;
firstWrittenAt: string;
lastWrittenAt: string;
/** When the last object needing this key expires; null when unknown. */
neededUntil: string | null;
}>;
/** The bucket's expiry, when it has one the credential may read. */
offhostExpirationDays: number | null;
/** Retired key ids some backup still needs (never the active one). */
retiredKeysStillNeeded: string[];
}
async function readBackupKeyNeeds(
activeKeyId: string,
now: Date,
): Promise<BackupKeyNeeds> {
const [stored, unrecorded, offhostRows] = await Promise.all([
prisma.$queryRaw<
Array<{ key_id: string; copies: number; oldest_at: Date }>
>`
SELECT k AS key_id, count(*)::int AS copies, min(created_at) AS oldest_at
FROM data_backups, unnest(inner_key_ids) AS k
WHERE inner_key_ids_recorded
GROUP BY k
ORDER BY k
`,
prisma.dataBackup.aggregate({
where: { innerKeyIdsRecorded: false },
_count: { _all: true },
_min: { createdAt: true },
}),
prisma.offhostBackupKeyUse.findMany({ orderBy: { keyId: "asc" } }),
]);
const lifecycle =
offhostRows.length > 0 && offhostBackupConfigured()
? await probeOffhostLifecycle()
: null;
const expirationDays =
lifecycle?.state === "configured" ? lifecycle.expirationDays : null;
const offhost = offhostRows.map((row) => ({
keyId: row.keyId,
firstWrittenAt: row.firstWrittenAt.toISOString(),
lastWrittenAt: row.lastWrittenAt.toISOString(),
neededUntil:
expirationDays === null
? null
: new Date(
row.lastWrittenAt.getTime() + (expirationDays + 1) * 86_400_000,
).toISOString(),
}));
const needed = new Set<string>();
for (const row of stored) needed.add(row.key_id);
for (const row of offhost) {
if (row.neededUntil === null || Date.parse(row.neededUntil) > +now) {
needed.add(row.keyId);
}
}
needed.delete(activeKeyId);
return {
stored: stored.map((row) => ({
keyId: row.key_id,
copies: row.copies,
oldestAt: row.oldest_at.toISOString(),
})),
unrecorded: {
copies: unrecorded._count._all,
oldestAt: unrecorded._min.createdAt?.toISOString() ?? null,
},
offhost,
offhostExpirationDays: expirationDays,
retiredKeysStillNeeded: [...needed].sort(),
};
}
export const GET = apiHandler(async () => {
await requireAdmin();
annotate({ action: { name: "admin.encryption.status" } });
@@ -114,6 +212,7 @@ export const GET = apiHandler(async () => {
}
const rotation = await readRotationState();
const backups = await readBackupKeyNeeds(scan.activeKeyId, new Date());
annotate({
meta: {
@@ -133,5 +232,14 @@ export const GET = apiHandler(async () => {
staleRows: scan.staleRows,
columns: scan.columns,
rotation,
backups,
// Every row rotated AND no backup still needs a retired key: the one
// state in which dropping an old key loses nothing. Copies that recorded
// no key ids count against it, because they may hold any key that
// existed when they were written.
safeToDropRetiredKeys:
scan.rotationComplete &&
backups.retiredKeysStillNeeded.length === 0 &&
backups.unrecorded.copies === 0,
});
});
@@ -25,6 +25,7 @@ vi.mock("next/headers", () => ({
}));
const mocks = vi.hoisted(() => ({
runningImportFindFirst: vi.fn(async () => null),
checkRateLimit: vi.fn().mockResolvedValue({ allowed: true }),
bossSend: vi.fn().mockResolvedValue("boss-job-1"),
getGlobalBoss: vi.fn(),
@@ -60,6 +61,16 @@ vi.mock("@/lib/db", () => ({
findFirst: mocks.importJobFindFirst,
update: mocks.importJobUpdate,
},
// The one-running-import check and the new row, in one transaction.
// The check finds nothing running unless a test says otherwise.
$transaction: async (fn: (tx: unknown) => unknown) =>
fn({
$executeRaw: async () => 0,
importJob: {
findFirst: mocks.runningImportFindFirst,
create: mocks.importJobCreate,
},
}),
user: {
findUnique: mocks.userFindUnique,
},
@@ -27,6 +27,11 @@ import {
type AppleHealthImportPayload,
} from "@/lib/jobs/apple-health-import-worker";
import { streamMultipartToDisk } from "@/lib/multipart/stream-to-disk";
import {
createImportJobUnlessBusy,
discardStagedUpload,
IMPORT_BUSY_CODE,
} from "@/lib/import/apple-health-staging";
import { unlink } from "node:fs/promises";
export const dynamic = "force-dynamic";
@@ -96,6 +101,7 @@ export const POST = apiHandler(async (request: NextRequest) => {
ipAddress: getClientIp(request),
details: { reason: "missing_user_id" },
});
await discardStagedUpload(uploaded.filePath);
return apiError("Multipart field 'userId' is required", 422);
}
@@ -109,6 +115,7 @@ export const POST = apiHandler(async (request: NextRequest) => {
ipAddress: getClientIp(request),
details: { reason: "target_user_missing", targetUserId },
});
await discardStagedUpload(uploaded.filePath);
return apiError(`Target user '${targetUserId}' does not exist`, 422);
}
@@ -149,19 +156,27 @@ export const POST = apiHandler(async (request: NextRequest) => {
ipAddress: getClientIp(request),
details: { reason: "worker_not_running" },
});
await discardStagedUpload(uploaded.filePath);
throw new HttpError(503, "Background worker is not running");
}
const importJob = await prisma.importJob.create({
data: {
userId: targetUser.id,
triggeredByAdminId: admin.id,
status: "queued",
uploadBytes: uploaded.bytes,
uploadSha256: uploaded.sha256,
parserRevision: APPLE_HEALTH_IMPORT_PARSER_REVISION,
},
const staged = await createImportJobUnlessBusy(prisma, {
userId: targetUser.id,
triggeredByAdminId: admin.id,
status: "queued",
uploadBytes: uploaded.bytes,
uploadSha256: uploaded.sha256,
parserRevision: APPLE_HEALTH_IMPORT_PARSER_REVISION,
});
if ("busy" in staged) {
await discardStagedUpload(uploaded.filePath);
return apiError(
"An Apple Health import is already running for this account. Wait for it to finish, then upload again.",
409,
{ errorCode: IMPORT_BUSY_CODE, jobId: staged.busy.id },
);
}
const importJob = staged.created;
const payload: AppleHealthImportPayload = {
userId: targetUser.id,
@@ -178,6 +193,19 @@ export const POST = apiHandler(async (request: NextRequest) => {
payload,
APPLE_HEALTH_IMPORT_SEND_OPTIONS,
);
if (!bossJobId) {
await prisma.importJob.update({
where: { id: importJob.id },
data: {
status: "failed",
failureReason:
"The import could not be queued. Upload the export again.",
completedAt: new Date(),
},
});
await discardStagedUpload(uploaded.filePath);
throw new HttpError(503, "The import could not be queued");
}
await prisma.importJob.update({
where: { id: importJob.id },
@@ -27,6 +27,7 @@ vi.mock("next/headers", () => ({
}));
const mocks = vi.hoisted(() => ({
runningImportFindFirst: vi.fn(async () => null),
checkRateLimit: vi.fn().mockResolvedValue({ allowed: true }),
bossSend: vi.fn().mockResolvedValue("boss-job-1"),
getGlobalBoss: vi.fn(),
@@ -35,6 +36,7 @@ const mocks = vi.hoisted(() => ({
importJobUpdate: vi.fn(),
streamToDisk: vi.fn(),
unlink: vi.fn().mockResolvedValue(undefined),
rm: vi.fn().mockResolvedValue(undefined),
}));
vi.mock("@/lib/rate-limit", () => ({
@@ -51,7 +53,7 @@ vi.mock("@/lib/multipart/stream-to-disk", () => ({
vi.mock("node:fs/promises", async (importOriginal) => {
const actual = await importOriginal<typeof import("node:fs/promises")>();
return { ...actual, unlink: mocks.unlink };
return { ...actual, unlink: mocks.unlink, rm: mocks.rm };
});
vi.mock("@/lib/db", () => ({
@@ -61,6 +63,16 @@ vi.mock("@/lib/db", () => ({
findFirst: mocks.importJobFindFirst,
update: mocks.importJobUpdate,
},
// The one-running-import check and the new row, in one transaction.
// The check finds nothing running unless a test says otherwise.
$transaction: async (fn: (tx: unknown) => unknown) =>
fn({
$executeRaw: async () => 0,
importJob: {
findFirst: mocks.runningImportFindFirst,
create: mocks.importJobCreate,
},
}),
},
}));
@@ -90,6 +102,7 @@ beforeEach(() => {
textFields: {},
});
mocks.unlink.mockResolvedValue(undefined);
mocks.rm.mockResolvedValue(undefined);
});
function multipartReq(): NextRequest {
@@ -220,3 +233,51 @@ describe("POST /api/import/apple-health-export — dedup by content hash (issue
);
});
});
// The staged upload is the person's export in plain text. Every exit that
// leaves no worker to consume it removes it, and a second import for the same
// account is refused while one is running.
describe("POST /api/import/apple-health-export — the staged upload on every exit", () => {
const removed = () =>
mocks.rm.mock.calls.some(([path]) => path === STAGED_PATH);
it("refuses a second import while one is running, and removes the upload", async () => {
vi.mocked(getSession).mockResolvedValue(SESSION_OK as never);
mocks.runningImportFindFirst.mockResolvedValue({
id: "ij-running",
status: "parsing",
} as never);
const res = await POST(multipartReq());
expect(res.status).toBe(409);
const body = await res.json();
expect(body.meta).toMatchObject({
errorCode: "import.apple_health.busy",
jobId: "ij-running",
});
expect(mocks.importJobCreate).not.toHaveBeenCalled();
expect(mocks.bossSend).not.toHaveBeenCalled();
expect(removed()).toBe(true);
});
it("removes the upload when no worker is running", async () => {
vi.mocked(getSession).mockResolvedValue(SESSION_OK as never);
mocks.getGlobalBoss.mockReturnValue(null);
const res = await POST(multipartReq());
expect(res.status).toBe(503);
expect(removed()).toBe(true);
});
it("fails the row and removes the upload when the queue takes no job", async () => {
vi.mocked(getSession).mockResolvedValue(SESSION_OK as never);
mocks.bossSend.mockResolvedValue(null);
const res = await POST(multipartReq());
expect(res.status).toBe(503);
expect(mocks.importJobUpdate).toHaveBeenCalledWith(
expect.objectContaining({
where: { id: "ij-1" },
data: expect.objectContaining({ status: "failed" }),
}),
);
expect(removed()).toBe(true);
});
});
@@ -35,6 +35,11 @@ import {
type AppleHealthImportPayload,
} from "@/lib/jobs/apple-health-import-worker";
import { streamMultipartToDisk } from "@/lib/multipart/stream-to-disk";
import {
createImportJobUnlessBusy,
discardStagedUpload,
IMPORT_BUSY_CODE,
} from "@/lib/import/apple-health-staging";
import { unlink } from "node:fs/promises";
export const dynamic = "force-dynamic";
@@ -147,18 +152,28 @@ export const POST = apiHandler(async (request: NextRequest) => {
ipAddress: getClientIp(request),
details: { reason: "worker_not_running" },
});
// Nothing will ever read the staged export; it must not stay in /tmp.
await discardStagedUpload(uploaded.filePath);
throw new HttpError(503, "Background worker is not running");
}
const importJob = await prisma.importJob.create({
data: {
userId: user.id,
status: "queued",
uploadBytes: uploaded.bytes,
uploadSha256: uploaded.sha256,
parserRevision: APPLE_HEALTH_IMPORT_PARSER_REVISION,
},
// One running import per account (see `apple-health-staging.ts`).
const staged = await createImportJobUnlessBusy(prisma, {
userId: user.id,
status: "queued",
uploadBytes: uploaded.bytes,
uploadSha256: uploaded.sha256,
parserRevision: APPLE_HEALTH_IMPORT_PARSER_REVISION,
});
if ("busy" in staged) {
await discardStagedUpload(uploaded.filePath);
return apiError(
"An Apple Health import is already running for this account. Wait for it to finish, then upload again.",
409,
{ errorCode: IMPORT_BUSY_CODE, jobId: staged.busy.id },
);
}
const importJob = staged.created;
const payload: AppleHealthImportPayload = {
userId: user.id,
@@ -174,6 +189,21 @@ export const POST = apiHandler(async (request: NextRequest) => {
payload,
APPLE_HEALTH_IMPORT_SEND_OPTIONS,
);
if (!bossJobId) {
// The queue did not take the job, so no worker will consume the upload
// or move the row on. Fail the row now and remove the export.
await prisma.importJob.update({
where: { id: importJob.id },
data: {
status: "failed",
failureReason:
"The import could not be queued. Upload the export again.",
completedAt: new Date(),
},
});
await discardStagedUpload(uploaded.filePath);
throw new HttpError(503, "The import could not be queued");
}
await prisma.importJob.update({
where: { id: importJob.id },
+10
View File
@@ -8,6 +8,10 @@ import {
MFA_STEP_UP_MAX_AGE_SECONDS,
} from "@/lib/api-handler";
import { annotate } from "@/lib/logging/context";
import {
kickOffhostPurge,
requestOffhostPurge,
} from "@/lib/jobs/offhost-purge";
import {
deleteGuardianAccountWithLifecycle,
LastManagedGuardianError,
@@ -101,6 +105,10 @@ export const DELETE = apiHandler(async (request: NextRequest) => {
// genuinely complete erasure.
await tx.auditLog.deleteMany({ where: { userId } });
// The nightly off-host copies go with the account. The request commits
// with the deletion; the purge job deletes the objects.
await requestOffhostPurge(tx, userId, "account_deleted");
// Delete user — all other related data is removed via onDelete: Cascade.
await tx.user.delete({ where: { id: userId } });
});
@@ -115,6 +123,8 @@ export const DELETE = apiHandler(async (request: NextRequest) => {
throw error;
}
await kickOffhostPurge();
annotate({
action: { name: "settings.account.delete" },
meta: { username },
+10
View File
@@ -1,5 +1,9 @@
import { prisma } from "@/lib/db";
import { auditLog } from "@/lib/auth/audit";
import {
kickOffhostPurge,
requestOffhostPurge,
} from "@/lib/jobs/offhost-purge";
import { apiSuccess, apiError, getClientIp } from "@/lib/api-response";
import { NextRequest } from "next/server";
import {
@@ -101,6 +105,11 @@ export const DELETE = apiHandler(async (request: NextRequest) => {
// does not. Same classification contract as the model list.
await tx.user.update({ where: { id: userId }, data: USER_RESET });
// The nightly off-host copies of the record are part of "all your
// data". The request commits with the wipe; the purge job deletes
// the objects and removes it.
await requestOffhostPurge(tx, userId, "data_wiped");
return perModel;
},
{
@@ -144,6 +153,7 @@ export const DELETE = apiHandler(async (request: NextRequest) => {
// v1.16.9 — every cached payload was built on the rows just deleted;
// hard-evict the user's buckets so no surface serves pre-wipe data.
invalidateUserData(userId);
await kickOffhostPurge();
return apiSuccess({ cleared: true, deletedRows, models: counts });
});
@@ -23,6 +23,11 @@ vi.mock("@tanstack/react-query", () => ({
import { I18nProvider } from "@/lib/i18n/context";
import { OffhostBackupsSection } from "../offhost-backups-section";
const QUIET = {
pendingDeletions: { count: 0, oldestRequestedAt: null, lastFailure: null },
lifecycle: { state: "configured" as const, expirationDays: 30 },
};
function render(data: BackupsList["offhost"] | null): string {
queryResult.current = {
data:
@@ -46,7 +51,12 @@ function render(data: BackupsList["offhost"] | null): string {
describe("<OffhostBackupsSection>", () => {
it("says off-host backup is not configured rather than showing an empty list", () => {
const html = render({ configured: false, periodHours: 24, rows: [] });
const html = render({
configured: false,
periodHours: 24,
rows: [],
...QUIET,
});
expect(html).toContain("Off-host backup is not configured");
expect(html).not.toContain('data-slot="offhost-backup-rows"');
});
@@ -55,6 +65,7 @@ describe("<OffhostBackupsSection>", () => {
const html = render({
configured: true,
periodHours: 24,
...QUIET,
rows: [
{
userId: "u1",
@@ -109,4 +120,33 @@ describe("<OffhostBackupsSection>", () => {
// Never the bucket's own coordinates.
expect(html).not.toMatch(/secret|access[- ]key/i);
});
it("says when the bucket keeps copies forever, and when deletions are stuck", () => {
const html = render({
configured: true,
periodHours: 24,
rows: [],
pendingDeletions: {
count: 2,
oldestRequestedAt: "2026-09-20T10:00:00.000Z",
lastFailure: "AccessDenied",
},
lifecycle: { state: "missing", expirationDays: null },
});
expect(html).toContain('data-lifecycle-state="missing"');
expect(html).toContain("no lifecycle rule");
expect(html).toContain('data-slot="offhost-pending-deletions"');
expect(html).toContain("AccessDenied");
});
it("states the expiry the bucket enforces", () => {
const html = render({
configured: true,
periodHours: 24,
rows: [],
...QUIET,
});
expect(html).toContain("after 30 days");
expect(html).not.toContain('data-slot="offhost-pending-deletions"');
});
});
+10 -1
View File
@@ -71,6 +71,7 @@ function restoredCount(job: BackupRestoreJobView): number {
function failureText(
code: BackupRestoreFailureCode,
t: ReturnType<typeof useTranslations>["t"],
keyIds: readonly string[] = [],
): string {
switch (code) {
case "backup_not_found":
@@ -91,6 +92,12 @@ function failureText(
return t("admin.section.backups.restoreFailureSectionMissing");
case "document_ciphertext_missing":
return t("admin.section.backups.restoreFailureDocument");
case "backup.key.missing":
return t("admin.section.backups.restoreFailureKeyMissing", {
keys: keyIds.join(", "),
});
case "backup.foreign_reference":
return t("admin.section.backups.restoreFailureForeignReference");
case "time_budget":
return t("admin.section.backups.restoreFailureTimeBudget");
case "transaction_failed":
@@ -289,7 +296,9 @@ export function RestoreJobStatus({
<p className="font-medium">
{t("admin.section.backups.restoreJobFailedTitle", { username })}
</p>
<p className="text-xs">{failureText(code, t)}</p>
<p className="text-xs">
{failureText(code, t, job.failure?.keyIds)}
</p>
{changedNothing ? (
<p className="text-muted-foreground mt-1 text-xs">
{t("admin.section.backups.restoreJobFailedKept")}
+17 -5
View File
@@ -238,11 +238,23 @@ function RestoreRowDialog({
{t("admin.section.backups.previewLoading")}
</p>
)}
{preview.isError && (
<p className="text-muted-foreground">
{t("admin.section.backups.previewUnavailable")}
</p>
)}
{preview.isError &&
(preview.error instanceof ApiError &&
preview.error.meta?.errorCode === "backup.key.missing" ? (
// The one preview failure the restore would repeat: say which
// key, so the operator can put it back before trying.
<p role="alert" className="text-destructive">
{t("admin.section.backups.previewKeyMissing", {
keys: Array.isArray(preview.error.meta.keyIds)
? preview.error.meta.keyIds.map(String).join(", ")
: "",
})}
</p>
) : (
<p className="text-muted-foreground">
{t("admin.section.backups.previewUnavailable")}
</p>
))}
{preview.isSuccess && (
<ul className="grid list-none grid-cols-1 gap-x-4 gap-y-0.5 p-0 sm:grid-cols-2">
{previewRows.map((entry) => (
+84 -3
View File
@@ -13,7 +13,13 @@
*/
import { useMutation, useQuery, useQueryClient } from "@tanstack/react-query";
import { KeyRound, Loader2, RotateCw, ShieldCheck } from "lucide-react";
import {
Archive,
KeyRound,
Loader2,
RotateCw,
ShieldCheck,
} from "lucide-react";
import { toast } from "sonner";
import {
AlertDialog,
@@ -32,7 +38,7 @@ import { StatTile } from "@/components/admin/_stat-tile";
import { Button } from "@/components/ui/button";
import { SettingsCard } from "@/components/settings/settings-card";
import { SettingsCardHeader } from "@/components/settings/_card-header";
import { useTranslations } from "@/lib/i18n/context";
import { useFormatters, useTranslations } from "@/lib/i18n/context";
import { queryKeys } from "@/lib/query-keys";
import { apiFetchRaw, apiGet } from "@/lib/api/api-fetch";
import { getApiErrorMessage } from "./_shared";
@@ -60,10 +66,25 @@ interface EncryptionStatus {
lastCompletedAt: string | null;
lastResult: { scanned: number; rotated: number; errors: number } | null;
};
/** Which keys the backups still need; see the status route. */
backups: {
stored: Array<{ keyId: string; copies: number; oldestAt: string }>;
unrecorded: { copies: number; oldestAt: string | null };
offhost: Array<{
keyId: string;
firstWrittenAt: string;
lastWrittenAt: string;
neededUntil: string | null;
}>;
offhostExpirationDays: number | null;
retiredKeysStillNeeded: string[];
};
safeToDropRetiredKeys: boolean;
}
export function EncryptionSection() {
const { t } = useTranslations();
const fmt = useFormatters();
const queryClient = useQueryClient();
const statusQuery = useQuery({
@@ -166,10 +187,20 @@ export function EncryptionSection() {
value={s.staleRows.toLocaleString()}
/>
</div>
{s.rotationComplete ? (
{s.safeToDropRetiredKeys ? (
<Badge className="border-success/40 bg-success/15 text-success">
{t("admin.section.encryption.safeToDropLegacy")}
</Badge>
) : s.rotationComplete ? (
// Every row is on the active key, and that is not the whole
// answer: a backup keeps the key its content was written under.
<Badge variant="secondary" data-slot="encryption-backups-need-keys">
{s.backups.retiredKeysStillNeeded.length > 0
? t("admin.section.encryption.backupsStillNeed", {
keys: s.backups.retiredKeysStillNeeded.join(", "),
})
: t("admin.section.encryption.backupsUnrecordedBadge")}
</Badge>
) : (
<Badge variant="secondary">
{t("admin.section.encryption.rotationIncomplete", {
@@ -179,6 +210,56 @@ export function EncryptionSection() {
)}
</SettingsCard>
{/* ── Keys the backups still need ───────────────────────────── */}
<SettingsCard>
<SettingsCardHeader
icon={Archive}
title={t("admin.section.encryption.backupsTitle")}
description={t("admin.section.encryption.backupsDescription")}
/>
<p className="text-sm">{t("admin.section.encryption.backupsDetail")}</p>
<ul className="space-y-1 text-sm" data-slot="encryption-backup-keys">
{s.backups.stored.map((row) => (
<li key={`stored-${row.keyId}`}>
{t("admin.section.encryption.backupsStoredLine", {
key: row.keyId,
count: row.copies,
when: fmt.dateTime(row.oldestAt),
})}
</li>
))}
{s.backups.unrecorded.copies > 0 && s.backups.unrecorded.oldestAt ? (
<li>
{t("admin.section.encryption.backupsUnrecordedLine", {
count: s.backups.unrecorded.copies,
when: fmt.dateTime(s.backups.unrecorded.oldestAt),
})}
</li>
) : null}
{s.backups.offhost.map((row) => (
<li key={`offhost-${row.keyId}`}>
{row.neededUntil
? t("admin.section.encryption.backupsOffhostLine", {
key: row.keyId,
when: fmt.dateTime(row.lastWrittenAt),
until: fmt.dateTime(row.neededUntil),
})
: t("admin.section.encryption.backupsOffhostLineUnknown", {
key: row.keyId,
when: fmt.dateTime(row.lastWrittenAt),
})}
</li>
))}
{s.backups.stored.length === 0 &&
s.backups.unrecorded.copies === 0 &&
s.backups.offhost.length === 0 ? (
<li className="text-muted-foreground">
{t("admin.section.encryption.backupsNone")}
</li>
) : null}
</ul>
</SettingsCard>
{/* ── Rotation status + trigger ──────────────────────────────── */}
<SettingsCard>
<SettingsCardHeader
@@ -11,11 +11,12 @@
* been refusing the worker's signature since March.
*
* Every verdict is computed server-side from the ledger the worker writes. The
* page never lists the bucket: the worker's grant is deliberately PutObject,
* GetObject and AbortMultipartUpload — the three the runbook documents, the
* abort being what the multipart path needs — and putting a credentialed
* listing call on the render path of an admin page would be a poor trade for
* a number the database already holds.
* page never lists the bucket: putting a credentialed listing call on the
* render path of an admin page would be a poor trade for a number the database
* already holds. It does read the bucket's lifecycle rule (one small call,
* cached for ten minutes on the server), because whether old copies expire at
* all is something only the bucket knows, and it states deletions of deleted
* or wiped accounts that the bucket has not carried out yet.
*/
import { useQuery } from "@tanstack/react-query";
@@ -191,6 +192,35 @@ export function OffhostBackupsSection() {
hours: offhost.periodHours,
})}
</p>
<p
className="text-xs"
data-slot="offhost-lifecycle"
data-lifecycle-state={offhost.lifecycle.state}
>
{offhost.lifecycle.state === "configured"
? t("admin.section.backups.offhost.lifecycleConfigured", {
days: offhost.lifecycle.expirationDays ?? 0,
})
: offhost.lifecycle.state === "missing"
? t("admin.section.backups.offhost.lifecycleMissing")
: t("admin.section.backups.offhost.lifecycleUnknown")}
</p>
{offhost.pendingDeletions.count > 0 ? (
<p
role="status"
className="text-xs"
data-slot="offhost-pending-deletions"
>
{offhost.pendingDeletions.lastFailure
? t("admin.section.backups.offhost.pendingDeletionsFailing", {
count: offhost.pendingDeletions.count,
error: offhost.pendingDeletions.lastFailure,
})
: t("admin.section.backups.offhost.pendingDeletions", {
count: offhost.pendingDeletions.count,
})}
</p>
) : null}
</div>
) : null}
</SettingsCard>
+16 -3
View File
@@ -465,13 +465,21 @@ export interface RawStreamEncryptor {
final(): Buffer;
}
/** Open a byte-level streaming encryptor under an explicit 32-byte key. */
export function createRawStreamEncryptor(key: Buffer): RawStreamEncryptor {
/**
* Open a byte-level streaming encryptor under an explicit 32-byte key.
* `aad`, when given, is authenticated with the ciphertext and has to be given
* again, byte for byte, to decrypt it.
*/
export function createRawStreamEncryptor(
key: Buffer,
aad?: Buffer,
): RawStreamEncryptor {
if (key.byteLength !== 32) {
throw new Error("Stream encryption key must be 32 bytes");
}
const iv = randomBytes(IV_LENGTH);
const cipher = createCipheriv(ALGORITHM, key, iv);
if (aad) cipher.setAAD(aad);
let done = false;
return {
@@ -499,7 +507,11 @@ export function createRawStreamEncryptor(key: Buffer): RawStreamEncryptor {
* unverified plaintext to save a copy would trade the authentication for
* memory.
*/
export function decryptRawStream(packed: Buffer, key: Buffer): Buffer {
export function decryptRawStream(
packed: Buffer,
key: Buffer,
aad?: Buffer,
): Buffer {
if (packed.byteLength < IV_LENGTH + AUTH_TAG_LENGTH) {
throw new Error("Streamed ciphertext is truncated");
}
@@ -508,6 +520,7 @@ export function decryptRawStream(packed: Buffer, key: Buffer): Buffer {
const ct = packed.subarray(IV_LENGTH, packed.byteLength - AUTH_TAG_LENGTH);
const dec = createDecipheriv(ALGORITHM, key, iv);
dec.setAuthTag(tag);
if (aad) dec.setAAD(aad);
return Buffer.concat([dec.update(ct), dec.final()]);
}
+7 -3
View File
@@ -227,9 +227,9 @@ export const WIPE_MODELS = [
// ended. About this account's record, so they go with it.
"BackupRestoreJob",
// What this host last put in the operator's bucket for this account. The
// objects themselves stay — the worker holds no DeleteObject grant and the
// bucket's lifecycle rule owns their retirement — but the row saying when
// they were written is about this account and goes with it.
// objects themselves are deleted by the purge request the wipe writes
// (`requestOffhostPurge`); this row, saying when they were written, is about
// this account and goes with it.
"OffhostBackupState",
"AuditLog",
"ConsentReceipt",
@@ -297,6 +297,10 @@ export const INSTANCE_SCOPED: Readonly<Record<string, string>> = {
AppSettings: "instance-wide configuration singleton",
RateLimit: "instance-wide rate-limit buckets, keyed by string, self-expiring",
HostMetric: "instance-wide host telemetry, not attributable to an account",
OffhostBackupKeyUse:
"per encryption key, when the nightly off-host run last wrote an object needing it; says nothing about any account",
OffhostPurgeRequest:
"the deletions an account deletion or wipe asked of the off-host bucket; written by the wipe itself and removed once the bucket holds nothing of the account",
CycleSymptomCategory:
"seeded catalogue shared by every account; a user's custom symptoms live on CycleSymptom",
IllnessSymptom: "seeded symptom catalogue shared by every account",
@@ -0,0 +1,178 @@
/**
* The key ids a backup's inner ciphertext needs, and the verdict against the
* keys this host has.
*
* The values are produced by the real codecs under real keys, then the host's
* key set is changed underneath them, which is exactly what an operator does
* when a rotation is followed by dropping the old key.
*/
import { Buffer } from "node:buffer";
import { afterEach, beforeEach, describe, expect, it } from "vitest";
import { _resetCryptoCacheForTests, encrypt, encryptBytes } from "@/lib/crypto";
import {
assessBackupKeys,
BackupKeyIdCollector,
createBackupKeyIdTextScanner,
describeBackupKeyProblem,
innerCiphertextKeyId,
} from "../backup-key-ids";
const K1 = "11".repeat(32);
const K2 = "22".repeat(32);
const K3 = "33".repeat(32);
function useKeys(keys: Record<string, string>, active: string) {
delete process.env.ENCRYPTION_KEY;
process.env.ENCRYPTION_KEYS = JSON.stringify(keys);
process.env.ENCRYPTION_ACTIVE_KEY_ID = active;
_resetCryptoCacheForTests();
}
const saved = {
key: process.env.ENCRYPTION_KEY,
keys: process.env.ENCRYPTION_KEYS,
active: process.env.ENCRYPTION_ACTIVE_KEY_ID,
};
beforeEach(() => useKeys({ old: K1, cur: K2 }, "old"));
afterEach(() => {
for (const [name, value] of [
["ENCRYPTION_KEY", saved.key],
["ENCRYPTION_KEYS", saved.keys],
["ENCRYPTION_ACTIVE_KEY_ID", saved.active],
] as const) {
if (value === undefined) delete process.env[name];
else process.env[name] = value;
}
_resetCryptoCacheForTests();
});
/** A Bytes column holding the string codec, as a DR payload carries it. */
const bytesString = (plain: string) =>
Buffer.from(encrypt(plain), "utf8").toString("base64");
/** A Bytes column holding the binary codec, as a document's content. */
const binary = (plain: string) =>
encryptBytes(Buffer.from(plain)).toString("base64");
describe("innerCiphertextKeyId", () => {
it("reads the key id of each stored layout", () => {
expect(innerCiphertextKeyId(encrypt("note"))).toEqual({
keyId: "old",
form: "string",
});
expect(innerCiphertextKeyId(bytesString("note"))).toEqual({
keyId: "old",
form: "bytes-string",
});
expect(innerCiphertextKeyId(binary("document"))).toEqual({
keyId: "old",
form: "binary",
});
});
it("reads a value with no key id as the legacy v1 key", () => {
const legacy = Buffer.from("A".repeat(40)).toString("base64");
expect(innerCiphertextKeyId(legacy)?.keyId).toBe("v1");
});
it("ignores an empty value, a short one, and one that is not base64", () => {
expect(innerCiphertextKeyId("")).toBeNull();
expect(innerCiphertextKeyId("yes")).toBeNull();
expect(
innerCiphertextKeyId("not ciphertext at all, and long enough to count!"),
).toBeNull();
});
});
describe("assessBackupKeys", () => {
function fileWrittenUnderOld() {
return {
userId: "u1",
moodEntries: [{ noteEncrypted: bytesString("a mood note") }],
documents: [{ contentEncrypted: binary("%PDF-1.4 …") }],
appSettings: { adminAiKeyEncrypted: encrypt("sk-something") },
};
}
it("passes a file whose keys are all configured and open", () => {
const collector = new BackupKeyIdCollector();
collector.visit(fileWrittenUnderOld());
collector.visit({ notesEncrypted: bytesString("x") }, "measurements");
const verdict = assessBackupKeys(collector);
expect(verdict).toEqual({
keyIds: ["old"],
missing: [],
unreadable: [],
affectedValues: 0,
});
expect(describeBackupKeyProblem(verdict)).toBeNull();
});
it("names a key the file needs and the host no longer has", () => {
const file = fileWrittenUnderOld();
useKeys({ cur: K2 }, "cur");
const collector = new BackupKeyIdCollector();
collector.visit(file);
const verdict = assessBackupKeys(collector);
expect(verdict.missing).toEqual(["old"]);
expect(verdict.affectedValues).toBe(3);
expect(describeBackupKeyProblem(verdict)).toContain("'old'");
});
it("leaves out sections the caller will not write", () => {
useKeys({ old: K1, cur: K2 }, "cur");
const file = {
moodEntries: [{ noteEncrypted: bytesString("fine") }],
appSettings: {
adminAiKeyEncrypted: `gone.${Buffer.alloc(40).toString("base64")}`,
},
};
const collector = new BackupKeyIdCollector();
collector.visit(file);
expect(assessBackupKeys(collector).missing).toEqual(["gone"]);
expect(
assessBackupKeys(collector, { ignoreSections: new Set(["appSettings"]) })
.missing,
).toEqual([]);
});
it("catches an id that is configured with different key material", () => {
const file = fileWrittenUnderOld();
useKeys({ old: K3, cur: K2 }, "cur");
const collector = new BackupKeyIdCollector();
collector.visit(file);
const verdict = assessBackupKeys(collector);
expect(verdict.missing).toEqual([]);
expect(verdict.unreadable).toEqual(["old"]);
});
});
describe("createBackupKeyIdTextScanner", () => {
it("finds every key id in JSON fed in arbitrary pieces", () => {
const early = bytesString("written before the rotation");
useKeys({ old: K1, cur: K2 }, "cur");
const late = binary("written after it");
const json = JSON.stringify({
measurements: [{ id: "m1", notesEncrypted: early }],
documents: [{ contentEncrypted: late, note: "plain" }],
plain: "noteEncrypted",
});
for (const size of [1, 7, 33, 4096]) {
const scanner = createBackupKeyIdTextScanner();
for (let i = 0; i < json.length; i += size) {
scanner.feed(Buffer.from(json.slice(i, i + size), "utf8"));
}
expect(scanner.keyIds()).toEqual(["cur", "old"]);
}
});
it("reads pretty-printed JSON", () => {
const scanner = createBackupKeyIdTextScanner();
scanner.feed(
JSON.stringify({ a: { noteEncrypted: bytesString("x") } }, null, 2),
);
expect(scanner.keyIds()).toEqual(["old"]);
});
});
@@ -0,0 +1,70 @@
/**
* An uploaded backup is counted after the gunzip as well as before it.
* Mutation that must turn this red: drop the inflated count (the bomb reads
* through to the end).
*/
import { gzipSync } from "node:zlib";
import { describe, expect, it } from "vitest";
import {
BackupUploadDecodeError,
decodeBackupUpload,
} from "../backup-upload-decode";
async function drain(
bytes: Buffer,
limits: Parameters<typeof decodeBackupUpload>[1],
): Promise<Buffer> {
async function* source() {
for (let i = 0; i < bytes.length; i += 4096) {
yield bytes.subarray(i, i + 4096);
}
}
const out: Buffer[] = [];
for await (const chunk of decodeBackupUpload(source(), limits)) {
out.push(Buffer.from(chunk));
}
return Buffer.concat(out);
}
describe("decodeBackupUpload", () => {
it("passes plain JSON through and gunzips a gzip file", async () => {
const json = Buffer.from(JSON.stringify({ userId: "u1", n: [1, 2, 3] }));
expect(await drain(json, { compressedLimit: 1024 })).toEqual(json);
expect(await drain(gzipSync(json), { compressedLimit: 1024 })).toEqual(
json,
);
});
it("refuses a gzip that inflates far past its own size", async () => {
const bomb = gzipSync(Buffer.alloc(32 * 1024 * 1024));
expect(bomb.length).toBeLessThan(64 * 1024);
const refused = drain(bomb, {
compressedLimit: 1024 * 1024,
minAllowance: 1024 * 1024,
ratio: 100,
});
await expect(refused).rejects.toBeInstanceOf(BackupUploadDecodeError);
await expect(refused).rejects.toMatchObject({
status: 413,
reason: "inflated_size_exceeded",
});
});
it("refuses a file that is not valid gzip", async () => {
const broken = Buffer.concat([
Buffer.from([0x1f, 0x8b]),
Buffer.from("not gzip"),
]);
await expect(
drain(broken, { compressedLimit: 1024 }),
).rejects.toMatchObject({ status: 422, reason: "invalid_gzip" });
});
it("refuses more bytes than the upload limit", async () => {
await expect(
drain(Buffer.alloc(10_000, 0x20), { compressedLimit: 5_000 }),
).rejects.toMatchObject({ status: 413, reason: "file_size_exceeded" });
});
});
@@ -420,9 +420,15 @@ function recordingTx(
return new Proxy({} as Record<string, unknown>, {
// `$executeRaw` and friends: the restore's per-account lock, and the
// measurement insert, which answer with a row count.
// measurement insert, which answer with a row count. `$queryRaw` is the
// tenant check's read of the schema's foreign keys, which a mock has none
// of (the integration suite runs it against the real catalogue).
get: (_t, model: string) =>
model.startsWith("$") ? async () => 0 : delegate(model),
model === "$queryRaw"
? async () => []
: model.startsWith("$")
? async () => 0
: delegate(model),
});
}
+333
View File
@@ -0,0 +1,333 @@
/**
* Which application keys the ciphertext INSIDE a backup was written under.
*
* A disaster-recovery backup carries every encrypted column the way the
* database holds it: a note, a document, a coach message travel as the stored
* ciphertext, base64-encoded, and a restore writes that ciphertext back
* verbatim. The envelope around the whole copy is sealed separately, and key
* rotation re-seals that envelope, but it cannot reach inside it, and it
* cannot reach a copy in the off-host bucket or a file an operator keeps on a
* disk at all. So a backup taken before a rotation still needs the key the
* rotation retired, and a restore without that key writes rows back that no
* reader can open. Nothing in the restore used to notice: the envelope opened,
* the schema passed, and the damage showed up later as notes that would not
* decrypt.
*
* This module answers the question the restore has to ask first: which key
* ids does this file need, and are they all here. Every encrypted column in a
* backup payload is named `…Encrypted`, and each value is one of:
*
* - the string codec as stored in a text column: `<keyId>.<base64>`;
* - a Bytes column, base64-encoded, holding the string codec as UTF-8;
* - a Bytes column, base64-encoded, holding the binary codec
* (`0x02 | keyIdLen | keyId | iv | tag | ct`);
* - a legacy value with no key id, which only the `v1` key opens.
*
* Two readers share the classification: a walk over a parsed document (the
* restore, the preview, the upload and the restore drill already hold one),
* and a text scanner for a backup that is being written, which never exists
* as a document (`createBackupKeyIdTextScanner`).
*/
import { Buffer } from "node:buffer";
import { decrypt, decryptBytes, getConfiguredKeyIds } from "@/lib/crypto";
/** The key id a value with no key id prefix was written under. */
export const LEGACY_INNER_KEY_ID = "v1";
const KEY_ID = /^[A-Za-z0-9_-]{1,32}$/;
const VERSIONED = /^([A-Za-z0-9_-]{1,32})\.[A-Za-z0-9+/=]+$/;
const BASE64 = /^[A-Za-z0-9+/]+={0,2}$/;
const BASE64_TAIL = /^[A-Za-z0-9+/=]*$/;
const BYTES_CODEC_VERSION = 0x02;
const MIN_CIPHERTEXT_CHARS = 40;
/** How an inner value is laid out, which decides how a sample is opened. */
export type InnerCiphertextForm = "string" | "bytes-string" | "binary";
export interface InnerCiphertext {
keyId: string;
form: InnerCiphertextForm;
}
/**
* The key id one inner ciphertext value needs, or null for something that is
* not ciphertext at all (an empty string). Decodes only a prefix, so a
* multi-megabyte document costs the same as a note.
*/
export function innerCiphertextKeyId(value: string): InnerCiphertext | null {
// The shortest ciphertext either codec writes is 28 bytes (nonce and tag),
// 40 characters as base64: anything shorter is not ciphertext, and reading
// it as a legacy value would demand a key for a field that holds none.
if (value.length < MIN_CIPHERTEXT_CHARS) return null;
const versioned = VERSIONED.exec(value.slice(0, 64));
if (versioned && value.indexOf(".") === versioned[1].length) {
return { keyId: versioned[1], form: "string" };
}
// A Bytes column, base64-encoded. 48 characters decode to 36 bytes, which
// covers the longest header either codec can have (2 + 32 for the binary
// one, 33 for `<keyId>.`).
const head = value.slice(0, 48);
if (!BASE64.test(head)) return null;
const bytes = Buffer.from(head, "base64");
if (bytes.length >= 3 && bytes[0] === BYTES_CODEC_VERSION) {
const keyIdLen = bytes[1];
if (keyIdLen >= 1 && keyIdLen <= 32 && bytes.length >= 2 + keyIdLen) {
const keyId = bytes.subarray(2, 2 + keyIdLen).toString("latin1");
if (KEY_ID.test(keyId)) return { keyId, form: "binary" };
}
}
const text = bytes.toString("latin1");
const dot = text.indexOf(".");
// The rest of the prefix has to read as base64 too, or a legacy value's
// random bytes could pass for `<keyId>.` about once in a thousand values.
if (
dot > 0 &&
KEY_ID.test(text.slice(0, dot)) &&
BASE64_TAIL.test(text.slice(dot + 1))
) {
return { keyId: text.slice(0, dot), form: "bytes-string" };
}
// No key id anywhere: the legacy layout, which `decrypt` opens with `v1`.
return { keyId: LEGACY_INNER_KEY_ID, form: "bytes-string" };
}
/** Whether a member name holds inner ciphertext. */
export function isEncryptedMember(name: string): boolean {
return name.endsWith("Encrypted") && name.length > "Encrypted".length;
}
interface KeyUse {
/** How many values need this key. */
count: number;
/** The top-level sections they sit in (`measurements`, `appSettings`, …). */
sections: Set<string>;
/** The shortest value seen, kept to prove the key actually opens it. */
sample: { value: string; form: InnerCiphertextForm } | null;
}
/** A sample longer than this is not kept; opening it proves nothing more. */
const MAX_SAMPLE_CHARS = 64 * 1024;
/**
* Collects the key ids a backup's inner ciphertext needs, with the section
* each one appears in and one sample per key.
*/
export class BackupKeyIdCollector {
private readonly uses = new Map<string, KeyUse>();
/** Record one inner value found in `section`. */
add(section: string, value: string): void {
const found = innerCiphertextKeyId(value);
if (!found) return;
let use = this.uses.get(found.keyId);
if (!use) {
use = { count: 0, sections: new Set(), sample: null };
this.uses.set(found.keyId, use);
}
use.count += 1;
use.sections.add(section);
if (
value.length <= MAX_SAMPLE_CHARS &&
(!use.sample || value.length < use.sample.value.length)
) {
use.sample = { value, form: found.form };
}
}
/**
* Walk a parsed value and record every `…Encrypted` string in it. `section`
* names where it sits in the file; for a whole document, pass nothing and
* each top-level member is its own section.
*/
visit(value: unknown, section?: string): void {
if (section === undefined) {
if (value && typeof value === "object" && !Array.isArray(value)) {
for (const [key, child] of Object.entries(value)) {
this.walk(child, key, key);
}
}
return;
}
this.walk(value, section, null);
}
private walk(value: unknown, section: string, member: string | null): void {
if (typeof value === "string") {
if (member !== null && isEncryptedMember(member)) {
this.add(section, value);
}
return;
}
if (Array.isArray(value)) {
// Elements of an array under an `…Encrypted` member are ciphertext too.
for (const element of value) this.walk(element, section, member);
return;
}
if (value && typeof value === "object") {
for (const [key, child] of Object.entries(value)) {
this.walk(child, section, key);
}
}
}
/** Every key id seen, sorted. */
keyIds(): string[] {
return [...this.uses.keys()].sort();
}
/** How many values need `keyId`. */
count(keyId: string): number {
return this.uses.get(keyId)?.count ?? 0;
}
/** @internal for `assessBackupKeys`. */
entries(): ReadonlyMap<string, KeyUse> {
return this.uses;
}
}
export interface BackupKeyAssessment {
/** Every key id the file's inner ciphertext was written under. */
keyIds: string[];
/** Needed and not in `ENCRYPTION_KEYS` on this host. */
missing: string[];
/**
* Configured, but the sample written under it does not open: the id is
* there with different key material behind it.
*/
unreadable: string[];
/** Values that need a missing or unreadable key. */
affectedValues: number;
}
function openSample(sample: { value: string; form: InnerCiphertextForm }) {
switch (sample.form) {
case "string":
decrypt(sample.value);
return;
case "bytes-string":
decrypt(Buffer.from(sample.value, "base64").toString("utf8"));
return;
case "binary":
decryptBytes(Buffer.from(sample.value, "base64"));
return;
}
}
/**
* Hold what the file needs against what this host has, and open one value per
* key to prove the key under that id is the right one.
*
* `ignoreSections` leaves out sections the caller will not write: the
* instance settings, when the operator has not asked for them back.
*/
export function assessBackupKeys(
collector: BackupKeyIdCollector,
options: { ignoreSections?: ReadonlySet<string> } = {},
): BackupKeyAssessment {
const ignore = options.ignoreSections ?? new Set<string>();
const keyIds: string[] = [];
const missing: string[] = [];
const unreadable: string[] = [];
let affectedValues = 0;
// Read lazily: a file with no inner ciphertext needs no key, and must not
// fail on a host (or a test) that has none configured.
let configured: Set<string> | null = null;
for (const [keyId, use] of [...collector.entries()].sort(([a], [b]) =>
a.localeCompare(b),
)) {
if ([...use.sections].every((section) => ignore.has(section))) continue;
keyIds.push(keyId);
configured ??= new Set(getConfiguredKeyIds());
if (!configured.has(keyId)) {
missing.push(keyId);
affectedValues += use.count;
continue;
}
if (!use.sample) continue;
try {
openSample(use.sample);
} catch {
unreadable.push(keyId);
affectedValues += use.count;
}
}
return { keyIds, missing, unreadable, affectedValues };
}
/** The refusal code every surface answers a key problem with. */
export const BACKUP_KEY_MISSING_CODE = "backup.key.missing";
/** One sentence an operator can act on, naming the key ids. */
export function describeBackupKeyProblem(
assessment: BackupKeyAssessment,
): string | null {
const parts: string[] = [];
if (assessment.missing.length > 0) {
const ids = assessment.missing.map((id) => `'${id}'`).join(", ");
parts.push(
`This backup holds ${assessment.affectedValues} encrypted ${
assessment.affectedValues === 1 ? "value" : "values"
} written under encryption key ${ids}, which is not in ENCRYPTION_KEYS on this server. ` +
`Add ${assessment.missing.length === 1 ? "that key" : "those keys"} back to ENCRYPTION_KEYS and restart, then try again. Nothing was changed.`,
);
}
if (assessment.unreadable.length > 0) {
const ids = assessment.unreadable.map((id) => `'${id}'`).join(", ");
parts.push(
`Encryption key ${ids} is configured, but it does not open the values in this backup written under that id: ` +
`the key material behind the id differs from the one the backup was written with. Nothing was changed.`,
);
}
return parts.length > 0 ? parts.join(" ") : null;
}
/**
* Key ids found in backup JSON as it is written, for a copy that never exists
* as one document. Fed the text in pieces; a member split across two pieces is
* read once the rest of it arrives.
*/
export interface BackupKeyIdTextScanner {
feed(piece: string | Uint8Array): void;
keyIds(): string[];
}
const MEMBER_VALUE =
/"([A-Za-z0-9_]*Encrypted)"\s*:\s*"([A-Za-z0-9+/=._-]{0,64})(")?/g;
/** Enough of the previous piece to finish any member that straddles it. */
const CARRY_CHARS = 256;
export function createBackupKeyIdTextScanner(): BackupKeyIdTextScanner {
const found = new Set<string>();
let carry = "";
return {
feed(piece) {
// latin1, not utf-8: every character this looks for is ASCII, and a
// piece may end in the middle of a multi-byte character.
const text =
carry +
(typeof piece === "string"
? piece
: Buffer.from(
piece.buffer,
piece.byteOffset,
piece.byteLength,
).toString("latin1"));
MEMBER_VALUE.lastIndex = 0;
let match: RegExpExecArray | null;
while ((match = MEMBER_VALUE.exec(text)) !== null) {
const [, member, value, closed] = match;
if (!isEncryptedMember(member)) continue;
// A value cut off by the end of the piece is read again next time.
if (!closed && value.length < 64) continue;
const inner = innerCiphertextKeyId(value);
if (inner) found.add(inner.keyId);
}
carry = text.slice(-CARRY_CHARS);
},
keyIds() {
return [...found].sort();
},
};
}
+116
View File
@@ -0,0 +1,116 @@
/**
* An uploaded backup file's bytes, gunzipped when they are gzip, with both
* ends counted.
*
* The upload counted what arrived and nothing after the gunzip, so a few
* megabytes of gzip could inflate without limit. The JSON scanner keeps every
* section but the measurements in memory, which turned such a file into an
* out-of-memory kill of the whole app. The inflated bytes are now held to
* {@link INFLATION_RATIO} times the compressed bytes read so far (a real
* backup compresses about tenfold), never less than
* {@link MIN_INFLATED_ALLOWANCE}, and never more than {@link MAX_INFLATED_BYTES}.
*/
import { Readable } from "node:stream";
import { createGunzip } from "node:zlib";
/** How far an uploaded gzip may expand, measured on the bytes read so far. */
export const INFLATION_RATIO = 100;
/** Below this much output the ratio is not what protects the process. */
export const MIN_INFLATED_ALLOWANCE = 256 * 1024 * 1024;
/**
* The largest file an upload may inflate to. A disaster-recovery backup of
* 1.25 million readings is 662 MB of JSON; this leaves room for several times
* that record and stops a bomb long before memory does.
*/
export const MAX_INFLATED_BYTES = 4 * 1024 * 1024 * 1024;
/** Thrown for a file that is too large or not valid gzip. */
export class BackupUploadDecodeError extends Error {
constructor(
readonly status: 413 | 422,
message: string,
readonly reason:
"file_size_exceeded" | "inflated_size_exceeded" | "invalid_gzip",
readonly size?: number,
) {
super(message);
this.name = "BackupUploadDecodeError";
}
}
export interface DecodeLimits {
/** Cap on the bytes as they arrive. */
compressedLimit: number;
ratio?: number;
minAllowance?: number;
maxInflated?: number;
}
/** Bytes → bytes, gunzipped when the file starts with the gzip magic. */
export async function* decodeBackupUpload(
source: AsyncIterable<Uint8Array>,
limits: DecodeLimits,
): AsyncGenerator<Uint8Array> {
const ratio = limits.ratio ?? INFLATION_RATIO;
const minAllowance = limits.minAllowance ?? MIN_INFLATED_ALLOWANCE;
const maxInflated = limits.maxInflated ?? MAX_INFLATED_BYTES;
let seen = 0;
async function* counted() {
for await (const chunk of source) {
seen += chunk.byteLength;
if (seen > limits.compressedLimit) {
throw new BackupUploadDecodeError(
413,
`Upload exceeds ${Math.round(limits.compressedLimit / 1024 / 1024)} MB limit`,
"file_size_exceeded",
seen,
);
}
yield chunk;
}
}
const iterator = counted()[Symbol.asyncIterator]();
const first = await iterator.next();
if (first.done) return;
const rest = (async function* () {
yield first.value;
for (;;) {
const next = await iterator.next();
if (next.done) return;
yield next.value;
}
})();
const head = first.value;
if (!(head.byteLength >= 2 && head[0] === 0x1f && head[1] === 0x8b)) {
yield* rest;
return;
}
const gunzip = Readable.from(rest).pipe(createGunzip());
let inflated = 0;
try {
for await (const chunk of gunzip) {
inflated += (chunk as Buffer).byteLength;
const allowed = Math.min(
maxInflated,
Math.max(minAllowance, seen * ratio),
);
if (inflated > allowed) {
gunzip.destroy();
throw new BackupUploadDecodeError(
413,
`The uploaded file expands to more than ${Math.round(allowed / 1024 / 1024)} MB, which no backup this server wrote does. Nothing was stored.`,
"inflated_size_exceeded",
inflated,
);
}
yield chunk as Buffer;
}
} catch (err) {
if (err instanceof BackupUploadDecodeError) throw err;
throw new BackupUploadDecodeError(
422,
"Uploaded file is not valid gzip",
"invalid_gzip",
);
}
}
+69 -1
View File
@@ -47,6 +47,17 @@ import {
type BackupSource,
type StreamedBackup,
} from "@/lib/export/streamed-backup";
import {
assessBackupKeys,
BACKUP_KEY_MISSING_CODE,
describeBackupKeyProblem,
} from "@/lib/export/backup-key-ids";
import {
assertNoNewForeignReferences,
countForeignReferences,
ForeignReferenceError,
listTenantEdges,
} from "@/lib/export/tenant-boundary";
import { encryptNote } from "@/lib/crypto/note-cipher";
import { encryptToBytes } from "@/lib/ai/coach/bytes-codec";
import { encryptContextToBytes } from "@/lib/labs/biomarker-store";
@@ -240,6 +251,8 @@ export type RestoreFailureCode =
| "owner_not_found"
| "backup.section.missing"
| "document_ciphertext_missing"
| "backup.key.missing"
| "backup.foreign_reference"
| "time_budget"
| "transaction_failed"
| "interrupted"
@@ -475,6 +488,40 @@ export async function restoreBackup(
);
}
// The ciphertext inside the file is written back verbatim, so every key it
// was written under has to be on this host, and has to be the key that
// wrote it. A copy taken before a key rotation still needs the key the
// rotation retired: the rotation re-seals the stored copy's envelope but
// cannot reach inside it. Refused here, above the first delete, because the
// alternative is an account whose notes, documents and coach history come
// back as rows no reader can open. The instance settings count only when
// the operator asked for them back.
const keyVerdict = assessBackupKeys(streamed.keys, {
ignoreSections: restoreInstanceSettings
? new Set<string>()
: new Set(["appSettings"]),
});
const keyProblem = describeBackupKeyProblem(keyVerdict);
if (keyProblem) {
const keyIds = [...keyVerdict.missing, ...keyVerdict.unreadable];
await auditLog("admin.backups.restore.failed", {
userId: input.actorUserId,
ipAddress: input.ipAddress,
details: {
backupId: backup.id,
ownerId,
reason: "key_missing",
missingKeyIds: keyVerdict.missing,
unreadableKeyIds: keyVerdict.unreadable,
affectedValues: keyVerdict.affectedValues,
},
});
return refused(422, BACKUP_KEY_MISSING_CODE, keyProblem, {
errorCode: BACKUP_KEY_MISSING_CODE,
keyIds,
});
}
// Audit the *intent* before the transaction begins so the trail
// describing "an admin is about to restore <user>" survives even
// if the operation crashes midway.
@@ -605,6 +652,17 @@ export async function restoreBackup(
// commit or roll back, then replaces the account again from its own
// file, which is what it was asked to do.
await tx.$executeRaw`SELECT pg_advisory_xact_lock(hashtext(${`backup-restore:${ownerId}`}))`;
// Every reference this restore writes stays inside the account: the
// references that leave it are counted now and again after the last
// write, and any that grew rolls the whole transaction back
// (`tenant-boundary.ts` says why the check is derived from the schema
// rather than written per section).
const tenantEdges = await listTenantEdges(tx);
const foreignBefore = await countForeignReferences(
tx,
ownerId,
tenantEdges,
);
// Declared INSIDE the transaction so a rollback takes the report with
// it. An accumulator that outlived a failed attempt would carry drops
// that never happened into the next one.
@@ -1835,7 +1893,7 @@ export async function restoreBackup(
for (const episode of payload.illnessEpisodes) {
if (episode.parentConditionId) {
await tx.illnessEpisode.update({
where: { id: episode.id },
where: { id: episode.id, userId: ownerId },
data: { parentConditionId: episode.parentConditionId },
});
}
@@ -2288,6 +2346,10 @@ export async function restoreBackup(
environmentCleared.environmentTravelLocations,
ecgRecordings: ecgCleared.ecgRecordings,
};
assertNoNewForeignReferences(
foreignBefore,
await countForeignReferences(tx, ownerId, tenantEdges),
);
progress.sectionsDone = RESTORE_SECTION_STEPS.length;
report("sections", true);
return { cleared, skipped: summarizeRestoreSkips(skips) };
@@ -2316,6 +2378,12 @@ export async function restoreBackup(
},
});
annotate({ meta: { restoreFailReason: verbose } });
if (err instanceof ForeignReferenceError) {
return refused(422, "backup.foreign_reference", err.message, {
errorCode: "backup.foreign_reference",
references: err.edges,
});
}
if (isStoredBackupReadError(err)) {
// The second read, inside the transaction, found the copy replaced or
// altered. The transaction rolled back, so nothing was changed.
+13 -1
View File
@@ -34,6 +34,7 @@ import {
type BackupJsonProducer,
type PackBackupOptions,
} from "@/lib/export/backup-blob";
import { createBackupKeyIdTextScanner } from "@/lib/export/backup-key-ids";
import {
BackupIntegrityError,
newChunkStreamId,
@@ -112,6 +113,11 @@ export async function storeBackupBlob(
await tx.dataBackupChunk.deleteMany({ where: { backupId: row.id } });
const streamId = newChunkStreamId();
// The key ids the copy's content was written under, read from the JSON
// on its way into the pieces. Rotation re-seals the pieces but never
// what is inside them, so this is what tells an operator how long a
// retired key is still needed (`GET /api/admin/encryption/status`).
const keyScanner = createBackupKeyIdTextScanner();
const { chunks, bytes } = await packBackupChunks(
async (sealed, seq) => {
await tx.dataBackupChunk.create({
@@ -120,7 +126,11 @@ export async function storeBackupBlob(
});
},
streamId,
producer,
(write) =>
producer(async (chunk) => {
keyScanner.feed(chunk);
await write(chunk);
}),
options,
);
@@ -135,6 +145,8 @@ export async function storeBackupBlob(
...(input.ownerAfterRead ? { userId: input.ownerAfterRead() } : {}),
chunkCount: chunks,
chunkStreamId: streamId,
innerKeyIds: keyScanner.keyIds(),
innerKeyIdsRecorded: true,
createdAt: new Date(),
},
});
+12
View File
@@ -21,6 +21,7 @@
* check reads). Run it through `parseBackupPayload` for the rest of the
* validation; together that is the whole-file check.
*/
import { BackupKeyIdCollector } from "@/lib/export/backup-key-ids";
import { scanBackupJson } from "@/lib/export/backup-json-scan";
import {
BACKUP_SCHEMA_VERSION,
@@ -51,6 +52,12 @@ export interface StreamedBackup {
measurementCount: number;
/** The file's size in bytes. */
bytes: number;
/**
* The application key ids the file's inner ciphertext was written under,
* gathered on the first read (`assessBackupKeys` holds them against this
* host's keys).
*/
keys: BackupKeyIdCollector;
/**
* Read the file again and hand its measurements over in file order, in
* batches of at most `size`, each row parsed by the element schema.
@@ -93,6 +100,7 @@ export async function readStreamedBackup(
): Promise<StreamedBackup> {
let firstWithoutId: number | null = null;
let bytes = 0;
const keys = new BackupKeyIdCollector();
async function* counted() {
for await (const chunk of source()) {
bytes += chunk.byteLength;
@@ -104,6 +112,7 @@ export async function readStreamedBackup(
onElement: (_key, element, index) => {
const row = parseMeasurement(element, index);
if (!row.id && firstWithoutId === null) firstWithoutId = index;
keys.visit(element, MEASUREMENTS);
options.onMeasurementChecked?.(index + 1);
},
});
@@ -120,6 +129,8 @@ export async function readStreamedBackup(
);
}
keys.visit(document);
const skipKeys = new Set(
Object.keys(document).filter((key) => key !== MEASUREMENTS),
);
@@ -128,6 +139,7 @@ export async function readStreamedBackup(
raw: document,
measurementCount: streamedCounts[MEASUREMENTS] ?? 0,
bytes,
keys,
async forEachMeasurementBatch(size, fn) {
let batch: BackupMeasurement[] = [];
await scanBackupJson(source(), {
+178
View File
@@ -0,0 +1,178 @@
/**
* Every reference a restore writes has to stay inside the account it writes.
*
* A backup file names ids: a custom mood tag names its category, an illness
* names its parent condition, a link names the document and the episode it
* joins. The restore writes those ids back as given, and a foreign key only
* proves the row they point at exists, not whose it is. So a file edited by
* hand, or an upload whose owner is simply what the file says it is, could
* attach the restored account's rows to rows of another account on the same
* host: a custom tag filed under somebody else's category, a note linked to
* somebody else's document. Several of the restore's writes (the tag and
* category upserts keyed on a globally unique `key`, for one) never even look
* at the owner.
*
* Checking each write site by hand is the approach that keeps missing one, so
* the check is derived from the schema instead. Every single-column foreign key
* is read from the database catalogue, and two shapes are held:
*
* - a table with `user_id` pointing at another table with `user_id`: a row of
* the restored account may not point at a row another account owns;
* - a link table without `user_id` joining two such tables: a link whose one
* end is the restored account's may not have its other end in another
* account.
*
* A NULL owner on the far end is a shared catalogue row (a seeded tag) and is
* allowed. The counts are taken twice inside the restore transaction, before
* the first delete and after the last write, and the restore is refused when
* any count went up: the host may have legitimate cross-account links of its
* own (a shared record), and the question is only whether THIS restore added
* one. Refusing throws, which rolls the whole transaction back.
*/
import type { Prisma } from "@/generated/prisma/client";
type Db = Pick<Prisma.TransactionClient, "$queryRawUnsafe" | "$queryRaw">;
/** One foreign key, or one pair of them through a link table. */
export interface TenantEdge {
/** Readable name for the refusal: `child.column -> parent`. */
name: string;
/** The counting query; `$1` is the owner id. */
sql: string;
}
/**
* Identifiers read from the catalogue are spliced into SQL, so each is held
* to this before use (the whitelist-splice rule for `$queryRawUnsafe`).
*/
const IDENT = /^[a-z_][a-z0-9_]*$/;
interface ForeignKeyRow {
child: string;
child_col: string;
parent: string;
parent_col: string;
child_owned: boolean;
parent_owned: boolean;
}
/** Thrown inside the restore transaction; rolls it back. */
export class ForeignReferenceError extends Error {
readonly edges: string[];
constructor(edges: string[]) {
super(
`The backup references records that belong to another account (${edges.join(", ")}). Nothing was changed.`,
);
this.name = "ForeignReferenceError";
this.edges = edges;
}
}
export async function listTenantEdges(db: Db): Promise<TenantEdge[]> {
const rows = await db.$queryRaw<ForeignKeyRow[]>`
SELECT
child.relname AS child,
ca.attname AS child_col,
parent.relname AS parent,
pa.attname AS parent_col,
EXISTS (
SELECT 1 FROM pg_attribute x
WHERE x.attrelid = c.conrelid AND x.attname = 'user_id'
AND NOT x.attisdropped
) AS child_owned,
EXISTS (
SELECT 1 FROM pg_attribute x
WHERE x.attrelid = c.confrelid AND x.attname = 'user_id'
AND NOT x.attisdropped
) AS parent_owned
FROM pg_constraint c
JOIN pg_class child ON child.oid = c.conrelid
JOIN pg_class parent ON parent.oid = c.confrelid
JOIN pg_namespace n ON n.oid = child.relnamespace
JOIN pg_attribute ca ON ca.attrelid = c.conrelid AND ca.attnum = c.conkey[1]
JOIN pg_attribute pa ON pa.attrelid = c.confrelid AND pa.attnum = c.confkey[1]
WHERE c.contype = 'f'
AND array_length(c.conkey, 1) = 1
AND n.nspname = current_schema()
ORDER BY child.relname, ca.attname
`;
const safe = rows.filter((row) =>
[row.child, row.child_col, row.parent, row.parent_col].every((id) =>
IDENT.test(id),
),
);
if (safe.length !== rows.length) {
// Never silently narrower than the schema: a check that quietly skips a
// table is the kind that stays green because it matched nothing.
throw new Error(
"Tenant check: a foreign key has an identifier outside [a-z0-9_]",
);
}
const edges: TenantEdge[] = [];
for (const fk of safe) {
if (!fk.child_owned || !fk.parent_owned) continue;
edges.push({
name: `${fk.child}.${fk.child_col} -> ${fk.parent}`,
sql:
`SELECT count(*)::int AS n FROM "${fk.child}" c ` +
`JOIN "${fk.parent}" p ON p."${fk.parent_col}" = c."${fk.child_col}" ` +
`WHERE c.user_id = $1 AND p.user_id IS NOT NULL AND p.user_id <> $1`,
});
}
// Link tables: no owner of their own, two or more keys into owned tables.
const byChild = new Map<string, ForeignKeyRow[]>();
for (const fk of safe) {
if (fk.child_owned || !fk.parent_owned) continue;
const list = byChild.get(fk.child) ?? [];
list.push(fk);
byChild.set(fk.child, list);
}
for (const [child, fks] of byChild) {
if (fks.length < 2) continue;
for (const near of fks) {
for (const far of fks) {
if (near === far) continue;
edges.push({
name: `${child}.${near.child_col} / ${child}.${far.child_col} -> ${far.parent}`,
sql:
`SELECT count(*)::int AS n FROM "${child}" l ` +
`JOIN "${near.parent}" a ON a."${near.parent_col}" = l."${near.child_col}" ` +
`JOIN "${far.parent}" b ON b."${far.parent_col}" = l."${far.child_col}" ` +
`WHERE a.user_id = $1 AND b.user_id IS NOT NULL AND b.user_id <> $1`,
});
}
}
}
return edges;
}
/** How many references from `ownerId`'s rows leave the account, per edge. */
export async function countForeignReferences(
db: Db,
ownerId: string,
edges: readonly TenantEdge[],
): Promise<Map<string, number>> {
const counts = new Map<string, number>();
for (const edge of edges) {
const [row] = await db.$queryRawUnsafe<Array<{ n: number }>>(
edge.sql,
ownerId,
);
counts.set(edge.name, row?.n ?? 0);
}
return counts;
}
/** Throw when any edge carries more foreign references than it did before. */
export function assertNoNewForeignReferences(
before: ReadonlyMap<string, number>,
after: ReadonlyMap<string, number>,
): void {
const grown = [...after]
.filter(([name, n]) => n > (before.get(name) ?? 0))
.map(([name]) => name);
if (grown.length > 0) throw new ForeignReferenceError(grown);
}
@@ -0,0 +1,52 @@
/**
* The sweep removes staged exports no running import can own, and nothing
* else. Mutations that must turn this red: drop the age check (the fresh
* upload of a running import goes), or widen the name pattern (a file the app
* did not write goes).
*/
import { mkdtempSync, readdirSync, utimesSync, writeFileSync } from "node:fs";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { describe, expect, it } from "vitest";
import {
STAGING_MAX_AGE_MS,
sweepStaleImportStaging,
} from "../apple-health-staging";
describe("sweepStaleImportStaging", () => {
it("removes old staged uploads and extracted XML, and leaves the rest", async () => {
const dir = mkdtempSync(join(tmpdir(), "healthlog-sweep-"));
const uuid = "0b7f3c2a-1d4e-4f5a-9b8c-7d6e5f4a3b2c";
const names = {
oldUpload: `healthlog-apple-health-import-${uuid}.bin`,
oldAdminUpload: `healthlog-admin-apple-health-import-${uuid}.bin`,
oldXml: `healthlog-import-${"ab".repeat(12)}.xml`,
freshUpload: `healthlog-upload-${uuid}.bin`,
foreign: "somebody-elses-file.bin",
};
for (const name of Object.values(names)) {
writeFileSync(join(dir, name), "x");
}
const now = Date.now();
const old = new Date(now - STAGING_MAX_AGE_MS - 60_000);
for (const name of [
names.oldUpload,
names.oldAdminUpload,
names.oldXml,
names.foreign,
]) {
utimesSync(join(dir, name), old, old);
}
expect(await sweepStaleImportStaging(dir, now)).toBe(3);
expect(readdirSync(dir).sort()).toEqual(
[names.foreign, names.freshUpload].sort(),
);
});
it("answers zero for a directory that is not there", async () => {
expect(await sweepStaleImportStaging("/nonexistent/healthlog")).toBe(0);
});
});
@@ -6,9 +6,12 @@ import { deflateRawSync } from "node:zlib";
import { pipeline } from "node:stream/promises";
import { Readable } from "node:stream";
import { readdirSync } from "node:fs";
import {
createByteCap,
extractExportXml,
memberOutputCap,
readCentralDirectory,
} from "../unzip-export-xml";
@@ -22,6 +25,8 @@ function buildMinimalZip(
filename: string,
payload: Buffer,
method: 0 | 8 = 8,
/** What the headers claim the member inflates to; the truth by default. */
declaredSize?: number,
): Buffer {
const compressed = method === 8 ? deflateRawSync(payload) : payload;
const crc32 = (() => {
@@ -54,7 +59,7 @@ function buildMinimalZip(
localHeader.writeUInt16LE(0, 12); // date
localHeader.writeUInt32LE(crc32, 14);
localHeader.writeUInt32LE(compressed.length, 18);
localHeader.writeUInt32LE(payload.length, 22);
localHeader.writeUInt32LE(declaredSize ?? payload.length, 22);
localHeader.writeUInt16LE(nameBuf.length, 26);
localHeader.writeUInt16LE(0, 28); // extra len
@@ -69,7 +74,7 @@ function buildMinimalZip(
cdh.writeUInt16LE(0, 14); // date
cdh.writeUInt32LE(crc32, 16);
cdh.writeUInt32LE(compressed.length, 20);
cdh.writeUInt32LE(payload.length, 24);
cdh.writeUInt32LE(declaredSize ?? payload.length, 24);
cdh.writeUInt16LE(nameBuf.length, 28);
cdh.writeUInt16LE(0, 30); // extra
cdh.writeUInt16LE(0, 32); // comment
@@ -166,6 +171,31 @@ describe("extractExportXml", () => {
expect(seenDeclaredBytes).toBe(payload.length);
});
// A member whose headers claim an honest ratio and whose deflate stream
// inflates a thousandfold. The header check passes it; only the count of
// the bytes actually produced, held to the compressed size on disk, stops
// it, and it must leave nothing behind in the staging directory.
it("refuses a member that inflates past its compressed size's cap, whatever it declares", async () => {
const payload = Buffer.alloc(80 * 1024 * 1024);
const compressedLength = deflateRawSync(payload).length;
const zip = buildMinimalZip(
"apple_health_export/export.xml",
payload,
8,
compressedLength * 2,
);
const tmp = mkdtempSync(join(tmpdir(), "healthlog-unzip-"));
const zipPath = join(tmp, "export.zip");
writeFileSync(zipPath, zip);
const staged = () =>
readdirSync(tmpdir()).filter((n) => n.startsWith("healthlog-import-"));
const before = new Set(staged());
await expect(extractExportXml(zipPath)).rejects.toThrow(/zip bomb/);
expect(staged().filter((n) => !before.has(n))).toEqual([]);
// Inflating past the 64 MiB floor takes a moment on a loaded runner.
}, 60_000);
it("throws when the export.xml member is missing", async () => {
const payload = Buffer.from("noop");
const zip = buildMinimalZip("other-file.txt", payload);
@@ -209,3 +239,26 @@ describe("createByteCap", () => {
).rejects.toThrow(/exceeds the 6-byte cap/);
});
});
describe("memberOutputCap", () => {
it("holds a deflated member to 200 times what it occupies, with a floor", () => {
expect(
memberOutputCap({ compressionMethod: 8, compressedSize: 1024 }),
).toBe(64 * 1024 * 1024);
expect(
memberOutputCap({
compressionMethod: 8,
compressedSize: 10 * 1024 * 1024,
}),
).toBe(2000 * 1024 * 1024);
expect(
memberOutputCap({ compressionMethod: 8, compressedSize: 1024 ** 3 }),
).toBe(8 * 1024 ** 3);
});
it("lets a stored member grow by nothing", () => {
expect(memberOutputCap({ compressionMethod: 0, compressedSize: 500 })).toBe(
500,
);
});
});
+114
View File
@@ -0,0 +1,114 @@
/**
* The staged Apple Health upload, from the request that writes it to the job
* that consumes it, and what happens to it on every way out.
*
* The upload is the person's whole health export in plain text, written to
* the shared temp directory by the kick-off request and read by the import
* worker. Until v1.39.3 several exits left it there: a worker that was not
* running (503), a queue that did not accept the job, an account deleted
* before the worker reached it, and nothing ever swept what was left. Every
* exit now removes it, and a periodic sweep removes anything older than the
* longest an import may run, whatever left it.
*
* It also holds the rule of one running import per account. An import holds
* the archive, the extracted XML (up to 8 GiB) and a worker slot; a second
* one for the same account only doubles that and races the first on the same
* rows. The check and the new row are one transaction under an advisory lock,
* so two uploads that arrive together cannot both pass.
*/
import { readdir, rm, stat } from "node:fs/promises";
import { tmpdir } from "node:os";
import { join } from "node:path";
import type { Prisma, PrismaClient } from "@/generated/prisma/client";
/** An import in one of these holds its staged files and a worker slot. */
export const ACTIVE_IMPORT_STATUSES = [
"queued",
"unpacking",
"parsing",
"upserting",
] as const;
/**
* An active row older than this is not holding anything any more: the queue
* expires an import after six hours (`APPLE_HEALTH_IMPORT_SEND_OPTIONS`) and
* the reconcile pass fails it soon after. It must not block the account.
*/
const ACTIVE_IMPORT_MAX_AGE_MS = 7 * 60 * 60 * 1000;
/** Staged files older than this belong to no running import. */
export const STAGING_MAX_AGE_MS = ACTIVE_IMPORT_MAX_AGE_MS;
/** The names the kick-off and the extractor give their temp files. */
const STAGING_NAME =
/^healthlog-(?:apple-health-import|admin-apple-health-import|upload)-[0-9a-f-]{36}\.bin$|^healthlog-import-[0-9a-f]{24}\.xml$/;
/** The error code a second concurrent import is refused with. */
export const IMPORT_BUSY_CODE = "import.apple_health.busy";
export type CreateImportJobResult =
{ created: { id: string } } | { busy: { id: string; status: string } };
/**
* Create the ImportJob row unless the account already has an import running.
* `data` is the row as the caller would create it.
*/
export async function createImportJobUnlessBusy(
prisma: PrismaClient,
data: Prisma.ImportJobUncheckedCreateInput,
now: Date = new Date(),
): Promise<CreateImportJobResult> {
return prisma.$transaction(async (tx) => {
await tx.$executeRaw`SELECT pg_advisory_xact_lock(hashtext(${`apple-health-import:${data.userId}`}))`;
const running = await tx.importJob.findFirst({
where: {
userId: data.userId,
status: { in: [...ACTIVE_IMPORT_STATUSES] },
startedAt: { gte: new Date(now.getTime() - ACTIVE_IMPORT_MAX_AGE_MS) },
},
orderBy: { startedAt: "desc" },
select: { id: true, status: true },
});
if (running) return { busy: running };
const created = await tx.importJob.create({ data, select: { id: true } });
return { created };
});
}
/** Remove a staged file. Never throws; a file already gone is fine. */
export async function discardStagedUpload(path: string): Promise<void> {
await rm(path, { force: true }).catch(() => {});
}
/**
* Remove staged uploads and extracted XML files older than
* `STAGING_MAX_AGE_MS` from `dir`. Only names this app writes are touched.
* Answers how many were removed.
*/
export async function sweepStaleImportStaging(
dir: string = tmpdir(),
now: number = Date.now(),
): Promise<number> {
let names: string[];
try {
names = await readdir(dir);
} catch {
return 0;
}
let removed = 0;
for (const name of names) {
if (!STAGING_NAME.test(name)) continue;
const path = join(dir, name);
try {
const info = await stat(path);
if (!info.isFile()) continue;
if (now - info.mtimeMs < STAGING_MAX_AGE_MS) continue;
await rm(path, { force: true });
removed++;
} catch {
// Gone between the listing and here, or not ours to remove.
}
}
return removed;
}
+34 -1
View File
@@ -73,6 +73,39 @@ const MAX_DECOMPRESSED_BYTES = 8 * 1024 * 1024 * 1024;
*/
const MAX_COMPRESSION_RATIO = 200;
/**
* Floor for the output cap of a small member. The ratio cap alone would
* refuse a tiny, legitimately very compressible export; below this many
* bytes of output the ratio is not what protects the disk.
*/
const MIN_OUTPUT_CAP_BYTES = 64 * 1024 * 1024;
/**
* How many bytes one member may inflate to, from what it ACTUALLY occupies
* in the archive rather than what its header claims. The header is the
* attacker's to write: a member declaring an honest-looking 10:1 ratio can
* still inflate at 1000:1, and with only the flat 8 GiB ceiling an 8 MB
* upload could fill `/tmp` with 8 GB before anything noticed. The streamed
* byte count is held to the ratio the preflight states, applied to the
* compressed bytes on disk, and never above the absolute ceiling. A stored
* member cannot grow at all.
*/
export function memberOutputCap(entry: {
compressionMethod: number;
compressedSize: number;
}): number {
if (entry.compressionMethod === 0) {
return Math.min(entry.compressedSize, MAX_DECOMPRESSED_BYTES);
}
return Math.min(
MAX_DECOMPRESSED_BYTES,
Math.max(
MIN_OUTPUT_CAP_BYTES,
entry.compressedSize * MAX_COMPRESSION_RATIO,
),
);
}
/**
* Cap on the central-directory slice the extractor is willing to hold
* in memory. A real Apple export's central directory is a few KB (one
@@ -636,7 +669,7 @@ async function streamEntryToFile(
}
const dest = createWriteStream(destPath);
const cap = createByteCap(MAX_DECOMPRESSED_BYTES);
const cap = createByteCap(memberOutputCap(entry));
try {
if (entry.compressedSize === 0) {
@@ -24,6 +24,10 @@ const mocks = vi.hoisted(() => ({
getJobById: vi.fn(),
}));
vi.mock("@/lib/import/apple-health-staging", () => ({
sweepStaleImportStaging: vi.fn(async () => 2),
}));
vi.mock("@/lib/db", () => ({
prisma: {
importJob: {
@@ -172,7 +176,7 @@ describe("handleImportJobReconcileTick", () => {
await expect(handleImportJobReconcileTick([] as never)).resolves.toEqual({
ok: true,
did: {},
did: { import_staging_swept: 2 },
});
expect(mocks.findMany).toHaveBeenCalledTimes(1);
});
@@ -37,6 +37,8 @@ const CODE_TABLE: Record<BackupRestoreFailureCode, true> = {
owner_not_found: true,
"backup.section.missing": true,
document_ciphertext_missing: true,
"backup.key.missing": true,
"backup.foreign_reference": true,
time_budget: true,
transaction_failed: true,
interrupted: true,
+119 -10
View File
@@ -32,6 +32,14 @@ import {
const ENC_KEY =
"0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef";
/** The purge ledger and key-use rows the run touches after each upload. */
function purgeLedgerMocks() {
return {
offhostPurgeRequest: { count: vi.fn().mockResolvedValue(0) },
offhostBackupKeyUse: { upsert: vi.fn() },
};
}
describe("offhost-backup envelope", () => {
it("encrypts and decrypts JSON round-trip", () => {
const key = Buffer.from(ENC_KEY, "hex");
@@ -184,8 +192,10 @@ describe("runOffhostBackup", () => {
s3.store.set("2020-01-01/user-old.json.enc", Buffer.from([0]));
const prisma = {
...purgeLedgerMocks(),
offhostBackupState: { upsert: vi.fn() },
user: {
count: vi.fn().mockResolvedValue(1),
findMany: vi.fn().mockResolvedValue([{ id: "u1" }, { id: "u2" }]),
},
measurement: { findMany: vi.fn().mockResolvedValue([]) },
@@ -214,7 +224,11 @@ describe("runOffhostBackup", () => {
expect(s3.deleteObject).not.toHaveBeenCalled();
const ct = s3.store.get("2026-05-08/user-u1.json.enc")!;
const decoded = decryptBackup(ct, Buffer.from(ENC_KEY, "hex"));
const decoded = decryptBackup(
ct,
Buffer.from(ENC_KEY, "hex"),
"2026-05-08/user-u1.json.enc",
);
const parsed = JSON.parse(decoded);
expect(parsed.userId).toBe("u1");
expect(() => backupPayloadSchema.parse(parsed)).not.toThrow();
@@ -248,11 +262,43 @@ describe("runOffhostBackup", () => {
expect(first.create.lastSuccessAt).toBeInstanceOf(Date);
});
it("takes back the copy of an account deleted while it was written", async () => {
const s3 = makeS3Mock();
const ledger = purgeLedgerMocks();
const prisma = {
...ledger,
offhostBackupState: { upsert: vi.fn() },
user: {
// u1 is gone by the time its upload lands; u2 is not.
count: vi.fn(async ({ where }: { where: { id: string } }) =>
where.id === "u1" ? 0 : 1,
),
findMany: vi.fn().mockResolvedValue([{ id: "u1" }, { id: "u2" }]),
},
};
const report = await runOffhostBackup(
// eslint-disable-next-line @typescript-eslint/no-explicit-any
prisma as any,
s3,
new Date("2026-05-08T03:00:00Z"),
);
expect(s3.store.has("2026-05-08/user-u1.json.enc")).toBe(false);
expect(s3.store.has("2026-05-08/user-u2.json.enc")).toBe(true);
expect(s3.deleteObject).toHaveBeenCalledWith("2026-05-08/user-u1.json.enc");
expect(report.uploaded).toBe(1);
// No ledger row for an account that no longer exists.
expect(prisma.offhostBackupState.upsert).toHaveBeenCalledTimes(1);
});
it("uploads the canonical builder output without reshaping it", async () => {
const s3 = makeS3Mock();
const prisma = {
...purgeLedgerMocks(),
offhostBackupState: { upsert: vi.fn() },
user: { findMany: vi.fn().mockResolvedValue([{ id: "u1" }]) },
user: {
count: vi.fn().mockResolvedValue(1),
findMany: vi.fn().mockResolvedValue([{ id: "u1" }]),
},
};
const canonicalPayload = {
schemaVersion: BACKUP_SCHEMA_VERSION,
@@ -301,7 +347,13 @@ describe("runOffhostBackup", () => {
expect(report.uploaded).toBe(1);
const ciphertext = s3.store.get("2026-05-08/user-u1.json.enc")!;
expect(
JSON.parse(decryptBackup(ciphertext, Buffer.from(ENC_KEY, "hex"))),
JSON.parse(
decryptBackup(
ciphertext,
Buffer.from(ENC_KEY, "hex"),
"2026-05-08/user-u1.json.enc",
),
),
).toEqual(canonicalPayload);
expect(() => backupPayloadSchema.parse(canonicalPayload)).not.toThrow();
});
@@ -310,8 +362,10 @@ describe("runOffhostBackup", () => {
const s3 = makeS3Mock();
const upsert = vi.fn();
const prisma = {
...purgeLedgerMocks(),
offhostBackupState: { upsert },
user: {
count: vi.fn().mockResolvedValue(1),
findMany: vi.fn().mockResolvedValue([{ id: "u1" }, { id: "u2" }]),
},
};
@@ -357,10 +411,14 @@ describe("runOffhostBackup", () => {
it("still reports a landed object as uploaded when the ledger write fails", async () => {
const s3 = makeS3Mock();
const prisma = {
...purgeLedgerMocks(),
offhostBackupState: {
upsert: vi.fn().mockRejectedValue(new Error("pool timeout")),
},
user: { findMany: vi.fn().mockResolvedValue([{ id: "u1" }]) },
user: {
count: vi.fn().mockResolvedValue(1),
findMany: vi.fn().mockResolvedValue([{ id: "u1" }]),
},
};
const report = await runOffhostBackup(
@@ -382,8 +440,12 @@ describe("runOffhostBackup", () => {
it("refuses an account whose object outgrows one multipart upload", async () => {
const s3 = makeS3Mock();
const prisma = {
...purgeLedgerMocks(),
offhostBackupState: { upsert: vi.fn() },
user: { findMany: vi.fn().mockResolvedValue([{ id: "u1" }]) },
user: {
count: vi.fn().mockResolvedValue(1),
findMany: vi.fn().mockResolvedValue([{ id: "u1" }]),
},
};
mocks.buildFullBackupPayload.mockResolvedValue({
payload: {
@@ -419,8 +481,12 @@ describe("runOffhostBackup", () => {
it("reports the largest object it wrote", async () => {
const s3 = makeS3Mock();
const prisma = {
...purgeLedgerMocks(),
offhostBackupState: { upsert: vi.fn() },
user: { findMany: vi.fn().mockResolvedValue([{ id: "u1" }]) },
user: {
count: vi.fn().mockResolvedValue(1),
findMany: vi.fn().mockResolvedValue([{ id: "u1" }]),
},
};
const report = await runOffhostBackup(
prisma as never,
@@ -453,8 +519,8 @@ describe("uploadEncryptedBackup", () => {
const stored = s3.store.get("k")!;
expect(stored.byteLength).toBe(bytes);
expect(stored.subarray(0, 5).toString("binary")).toBe("HLBK\x03");
expect(decryptBackup(stored, key)).toBe(document);
expect(stored.subarray(0, 5).toString("binary")).toBe("HLBK\x04");
expect(decryptBackup(stored, key, "k")).toBe(document);
});
it("restores an object of the old shape and one of the new one alike", async () => {
@@ -470,10 +536,53 @@ describe("uploadEncryptedBackup", () => {
const oldBytes = s3.store.get("old")!;
const newBytes = s3.store.get("new")!;
expect(oldBytes.subarray(0, 5).toString("binary")).toBe("HLBK\x02");
expect(newBytes.subarray(0, 5).toString("binary")).toBe("HLBK\x03");
expect(newBytes.subarray(0, 5).toString("binary")).toBe("HLBK\x04");
// Different framing, same record, one reader.
expect(decryptBackup(oldBytes, key)).toBe(document);
expect(decryptBackup(newBytes, key)).toBe(document);
expect(decryptBackup(newBytes, key, "new")).toBe(document);
});
it("binds a new object to its key in the bucket", async () => {
const s3 = makeS3Mock();
const document = JSON.stringify({ userId: "u1" });
await uploadEncryptedBackup(
s3,
"2026-09-01/user-u1.json.enc",
key,
(write) => write(document),
);
const stored = s3.store.get("2026-09-01/user-u1.json.enc")!;
expect(decryptBackup(stored, key, "2026-09-01/user-u1.json.enc")).toBe(
document,
);
// The same bytes presented as another account's copy, or another night's,
// do not open.
expect(() =>
decryptBackup(stored, key, "2026-09-01/user-u2.json.enc"),
).toThrow();
expect(() =>
decryptBackup(stored, key, "2026-09-02/user-u1.json.enc"),
).toThrow();
});
it("reads objects under a retired key through the previous-keys ring", async () => {
const s3 = makeS3Mock();
const document = JSON.stringify({ userId: "u1" });
const oldKey = Buffer.alloc(32, 7);
const newKey = Buffer.alloc(32, 9);
await uploadEncryptedBackup(s3, "k4", oldKey, (write) => write(document));
await s3.putObject("k2", encryptBackup(document, oldKey));
const ring = { active: newKey, previous: [oldKey] };
expect(decryptBackup(s3.store.get("k4")!, ring, "k4")).toBe(document);
expect(decryptBackup(s3.store.get("k2")!, ring)).toBe(document);
// Without the retired key the new-format object names the key it needs.
expect(() =>
decryptBackup(
s3.store.get("k4")!,
{ active: newKey, previous: [] },
"k4",
),
).toThrow(/BACKUP_ENCRYPTION_PREVIOUS_KEYS/);
});
it("rejects a tampered version-3 object rather than returning a partial one", async () => {
@@ -0,0 +1,189 @@
/**
* Off-host copies leave the bucket with the account.
*
* The bucket is a map behind the same `S3Like` boundary the real client
* implements; the request ledger is an in-memory table. Mutations that must
* turn this red: drop the `deleteObject` call (the objects stay), remove the
* request on failure too (the refusal case loses the request), or match on the
* key prefix instead of the full account pattern (another account's objects,
* or the health-check probe, go too).
*/
import { beforeEach, describe, expect, it, vi } from "vitest";
vi.mock("@/lib/auth/audit", () => ({ auditLog: vi.fn() }));
import { auditLog } from "@/lib/auth/audit";
import { processOffhostPurges, requestOffhostPurge } from "../offhost-purge";
interface Row {
id: string;
subjectId: string;
reason: string;
requestedAt: Date;
attempts: number;
lastAttemptAt: Date | null;
lastFailure: string | null;
}
function makeLedger(existingUsers: string[] = []) {
const rows: Row[] = [];
let seq = 0;
const prisma = {
offhostPurgeRequest: {
create: vi.fn(
async ({ data }: { data: { subjectId: string; reason: string } }) => {
const row: Row = {
id: `r${++seq}`,
subjectId: data.subjectId,
reason: data.reason,
requestedAt: new Date(Date.now() + seq),
attempts: 0,
lastAttemptAt: null,
lastFailure: null,
};
rows.push(row);
return { id: row.id };
},
),
findMany: vi.fn(async () =>
[...rows].sort((a, b) => +a.requestedAt - +b.requestedAt),
),
delete: vi.fn(async ({ where }: { where: { id: string } }) => {
rows.splice(
rows.findIndex((r) => r.id === where.id),
1,
);
}),
updateMany: vi.fn(
async ({
where,
data,
}: {
where: { id: { in: string[] } };
data: { lastAttemptAt: Date; lastFailure: string };
}) => {
for (const row of rows) {
if (!where.id.in.includes(row.id)) continue;
row.attempts += 1;
row.lastAttemptAt = data.lastAttemptAt;
row.lastFailure = data.lastFailure;
}
},
),
},
user: {
count: vi.fn(async ({ where }: { where: { id: string } }) =>
existingUsers.includes(where.id) ? 1 : 0,
),
},
};
return { rows, prisma };
}
function makeBucket(keys: string[]) {
const store = new Set(keys);
return {
store,
putObject: vi.fn(),
putStream: vi.fn(),
getObject: vi.fn(),
headObject: vi.fn(),
listObjects: vi.fn(async (prefix: string) =>
[...store].filter((k) => k.startsWith(prefix)).map((key) => ({ key })),
),
deleteObject: vi.fn(async (key: string) => {
store.delete(key);
}),
};
}
const BUCKET = [
"2026-09-01/user-gone.json.enc",
"2026-09-02/user-gone.json.enc",
"2026-09-02/user-gone2.json.enc",
"2026-09-02/user-stays.json.enc",
"_healthcheck/1.bin",
];
beforeEach(() => {
vi.unstubAllEnvs();
vi.clearAllMocks();
vi.stubEnv("BACKUP_S3_ENDPOINT", "https://r2.example");
vi.stubEnv("BACKUP_S3_BUCKET", "hl-backups");
vi.stubEnv("BACKUP_S3_ACCESS_KEY", "AKIA");
vi.stubEnv("BACKUP_S3_SECRET_KEY", "secret");
vi.stubEnv("BACKUP_ENCRYPTION_KEY", "ab".repeat(32));
});
describe("off-host purge", () => {
it("deletes every copy of the account, and nothing else", async () => {
const { rows, prisma } = makeLedger();
await requestOffhostPurge(prisma as never, "gone", "account_deleted");
const bucket = makeBucket(BUCKET);
const report = await processOffhostPurges(prisma as never, bucket);
expect([...bucket.store].sort()).toEqual([
"2026-09-02/user-gone2.json.enc",
"2026-09-02/user-stays.json.enc",
"_healthcheck/1.bin",
]);
expect(report).toMatchObject({
completed: 1,
failed: 0,
objectsDeleted: 2,
});
expect(rows).toHaveLength(0);
// The account is gone, and so is its audit trail: the receipt carries the
// reason and the count, not the account.
expect(auditLog).toHaveBeenCalledWith(
"offhost.backup.purged",
expect.objectContaining({
userId: null,
details: expect.objectContaining({
reason: "account_deleted",
objectsDeleted: 2,
}),
}),
);
});
it("keeps the request, with the bucket's answer, when a delete is refused", async () => {
const { rows, prisma } = makeLedger(["gone"]);
await requestOffhostPurge(prisma as never, "gone", "data_wiped");
const bucket = makeBucket(BUCKET);
bucket.deleteObject.mockRejectedValueOnce(new Error("AccessDenied"));
const report = await processOffhostPurges(prisma as never, bucket);
expect(report).toMatchObject({ completed: 0, failed: 1 });
expect(rows).toHaveLength(1);
expect(rows[0]).toMatchObject({ attempts: 1, lastFailure: "AccessDenied" });
// The next run finishes it, and the account that still exists keeps the
// receipt.
const again = await processOffhostPurges(prisma as never, bucket);
expect(again.completed).toBe(1);
expect(rows).toHaveLength(0);
expect(auditLog).toHaveBeenCalledWith(
"offhost.backup.purged",
expect.objectContaining({ userId: "gone" }),
);
});
it("records nothing on a host without off-host backup", async () => {
vi.stubEnv("BACKUP_S3_ENDPOINT", "");
const { rows, prisma } = makeLedger();
expect(
await requestOffhostPurge(prisma as never, "gone", "account_deleted"),
).toBe(false);
expect(rows).toHaveLength(0);
});
it("does not list the bucket when nothing is pending", async () => {
const { prisma } = makeLedger();
const bucket = makeBucket(BUCKET);
await processOffhostPurges(prisma as never, bucket);
expect(bucket.listObjects).not.toHaveBeenCalled();
});
});
@@ -1,5 +1,6 @@
import type { Readable } from "node:stream";
import { describe, it, expect, beforeEach, vi } from "vitest";
import { _resetCryptoCacheForTests, encrypt } from "@/lib/crypto";
import { encryptBackup } from "../offhost-backup";
import {
handleRestoreDrill,
@@ -115,6 +116,43 @@ describe("runRestoreDrill", () => {
expect(report.stale).toBe(true);
});
it("decrypts a value per inner key id and names a key the host lacks", async () => {
vi.stubEnv("ENCRYPTION_KEY", "");
vi.stubEnv(
"ENCRYPTION_KEYS",
JSON.stringify({ old: "11".repeat(32), cur: "22".repeat(32) }),
);
vi.stubEnv("ENCRYPTION_ACTIVE_KEY_ID", "old");
_resetCryptoCacheForTests();
const note = Buffer.from(encrypt("a note"), "utf8").toString("base64");
const s3 = makeS3Mock({
"2026-06-01/user-abc.json.enc": backupObject({
measurements: [{ id: "m1", notesEncrypted: note }],
}),
});
const now = new Date("2026-06-02T04:11:00Z");
const report = await runRestoreDrill(s3, now);
expect(report.innerKeyIds).toEqual(["old"]);
// The rotation ran, and the operator dropped the old key.
vi.stubEnv("ENCRYPTION_KEYS", JSON.stringify({ cur: "22".repeat(32) }));
vi.stubEnv("ENCRYPTION_ACTIVE_KEY_ID", "cur");
_resetCryptoCacheForTests();
await expect(runRestoreDrill(s3, now)).rejects.toThrow(/'old'/);
// Same id, different key material behind it.
vi.stubEnv(
"ENCRYPTION_KEYS",
JSON.stringify({ old: "33".repeat(32), cur: "22".repeat(32) }),
);
_resetCryptoCacheForTests();
await expect(runRestoreDrill(s3, now)).rejects.toThrow(
/does not open the values/,
);
vi.unstubAllEnvs();
_resetCryptoCacheForTests();
});
it("throws when the bucket holds no backup-shaped objects", async () => {
const s3 = makeS3Mock({ "_healthcheck/1.bin": Buffer.from([0x42]) });
await expect(runRestoreDrill(s3)).rejects.toThrow(/no backup objects/);
+50 -19
View File
@@ -49,6 +49,7 @@ import {
} from "@/lib/rollups/measurement-rollups";
import { withBackgroundEvent } from "@/lib/logging/background";
import { DEFAULT_TIMEZONE } from "@/lib/tz/format";
import { sweepStaleImportStaging } from "@/lib/import/apple-health-staging";
/**
* Queue + cron for the periodic orphan-ImportJob sweep. v1.32.1
@@ -90,13 +91,18 @@ export async function handleImportJobReconcileTick(
return withBackgroundEvent(
"job.apple_health_import_reconcile",
async (evt) => {
// Staged uploads and extracted XML that no running import can still
// own: whatever exit left them (a crash, a restart mid-import, a
// deploy), the person's export does not stay in /tmp past this.
const swept = await sweepStaleImportStaging().catch(() => 0);
evt.addMeta("import_staging_swept", swept);
try {
await reconcileOrphanImportJobs();
} catch (err) {
evt.addWarning(`apple-health-import-reconcile failed: ${err}`);
return jobFailed("apple health import reconcile failed", err);
}
return jobDone();
return jobDone({ import_staging_swept: swept });
},
);
}
@@ -327,16 +333,31 @@ export async function handleAppleHealthImport(
`[apple-health-import] No ImportJob row for pgBossJobId=${job.id};` +
" creating a stand-in",
);
importJob = await prisma.importJob.create({
data: {
userId,
triggeredByAdminId: triggeredByAdminId ?? null,
pgBossJobId: job.id,
status: "queued",
uploadBytes,
parserRevision: APPLE_HEALTH_IMPORT_PARSER_REVISION,
},
});
// The row is gone when the account is: an account deleted between the
// upload and this run takes its ImportJob with it by cascade. Then there
// is nobody to import for, and the staged export is removed rather than
// left in /tmp by a stand-in create that fails on the missing account.
const accountExists =
(await prisma.user.count({ where: { id: userId } })) > 0;
if (!accountExists) {
safeUnlink(uploadPath);
return jobDone({ skipped: "account_gone" });
}
try {
importJob = await prisma.importJob.create({
data: {
userId,
triggeredByAdminId: triggeredByAdminId ?? null,
pgBossJobId: job.id,
status: "queued",
uploadBytes,
parserRevision: APPLE_HEALTH_IMPORT_PARSER_REVISION,
},
});
} catch (err) {
safeUnlink(uploadPath);
throw err;
}
}
const importJobId = importJob.id;
@@ -588,16 +609,26 @@ export async function handleAppleHealthImport(
: err instanceof Error
? err.message
: String(err);
await prisma.importJob.update({
where: { id: importJobId },
data: {
status: "failed",
failureReason: reason.slice(0, 1000),
completedAt: new Date(),
},
});
// The files first: the row update below throws when the account was
// deleted mid-import (the row went with it), and the export must not be
// left behind in that case either.
if (extractedXmlPath) safeUnlink(extractedXmlPath);
safeUnlink(uploadPath);
try {
await prisma.importJob.update({
where: { id: importJobId },
data: {
status: "failed",
failureReason: reason.slice(0, 1000),
completedAt: new Date(),
},
});
} catch (updateErr) {
console.warn(
`[apple-health-import] Could not record the failure on ImportJob=${importJobId}`,
updateErr,
);
}
throw err;
}
}
+8
View File
@@ -120,12 +120,15 @@ export interface BackupRestoreFailure {
message: string;
/** The sections a file's manifest names and it does not carry. */
sections?: string[];
/** The encryption key ids a file needs and this host cannot open it with. */
keyIds?: string[];
}
const failureSchema = z.object({
code: z.string(),
message: z.string(),
sections: z.array(z.string()).optional(),
keyIds: z.array(z.string()).optional(),
});
const progressSchema = z.object({
@@ -308,6 +311,7 @@ function failureJson(failure: BackupRestoreFailure) {
code: failure.code,
message: failure.message,
...(failure.sections ? { sections: failure.sections } : {}),
...(failure.keyIds ? { keyIds: failure.keyIds } : {}),
});
}
@@ -632,11 +636,15 @@ export async function runBackupRestoreJob(
const sections = Array.isArray(outcome.meta?.sections)
? (outcome.meta.sections as unknown[]).map(String)
: undefined;
const keyIds = Array.isArray(outcome.meta?.keyIds)
? (outcome.meta.keyIds as unknown[]).map(String)
: undefined;
await finish("failed", {
failure: {
code: outcome.code,
message: outcome.message,
...(sections ? { sections } : {}),
...(keyIds ? { keyIds } : {}),
},
});
// A file refused by its checks is the operator's to fix and fails the
+5
View File
@@ -107,6 +107,7 @@ export const JOB_FACT_ALLOWLIST: ReadonlySet<string> = new Set([
"geolite2_fetch_status",
"host_metric_pruned",
"hourly_rows_upserted",
"import_staging_swept",
"imported",
"in_slot",
"in_window",
@@ -144,6 +145,10 @@ export const JOB_FACT_ALLOWLIST: ReadonlySet<string> = new Set([
"offhost_backup_oversized",
"offhost_backup_total_users",
"offhost_backup_uploaded",
"offhost_purge_completed",
"offhost_purge_failed",
"offhost_purge_objects_deleted",
"offhost_purge_pending",
"outcome",
"passes_failed",
"per_sample_rows_deleted",
+345 -55
View File
@@ -11,23 +11,29 @@
* Object key layout:
* <bucket>/<YYYY-MM-DD>/user-<userId>.json.enc
*
* Retention: the worker NEVER calls DeleteObject on backup keys. Operators
* MUST configure a bucket-level lifecycle rule (e.g. expire after
* `BACKUP_RETENTION_DAYS`, which this module reads nowhere — see
* `loadOffhostConfig`). This keeps the IAM grant for the worker
* limited to PutObject + GetObject + AbortMultipartUpload, so a compromised
* worker cannot wipe the backup history. The abort is what cleans up a run
* that failed partway rather than leaving billed, unlistable parts behind;
* `AbortMultipartUpload` can only touch an upload this worker started, never
* a finished object. See docs/ops/backup-restore.md.
* Retention: the worker does not expire objects by age. Operators configure a
* bucket-level lifecycle rule (e.g. expire after `BACKUP_RETENTION_DAYS`,
* which this module reads nowhere — see `loadOffhostConfig`); the admin
* off-host card reads that rule back (`probeOffhostLifecycle`) and says when
* there is none. Objects ARE deleted in one case: an account that is deleted
* or whose data is wiped (`offhost-purge.ts`), which needs DeleteObject in the
* grant. `AbortMultipartUpload` cleans up a run that failed partway rather
* than leaving billed, unlistable parts behind. See
* docs/ops/backup-restore.md.
*/
import { envOr } from "@/lib/env";
import { Buffer } from "node:buffer";
import { createCipheriv, createDecipheriv, randomBytes } from "node:crypto";
import {
createCipheriv,
createDecipheriv,
createHash,
randomBytes,
} from "node:crypto";
import { Readable, Transform } from "node:stream";
import { createGunzip, createGzip, gunzipSync, gzipSync } from "node:zlib";
import type { PrismaClient } from "@/generated/prisma/client";
import { createRawStreamEncryptor, decryptRawStream } from "@/lib/crypto";
import { createBackupKeyIdTextScanner } from "@/lib/export/backup-key-ids";
import { streamFullBackupJson } from "@/lib/export/full-backup-stream";
import { annotate, getEvent } from "@/lib/logging/context";
@@ -41,7 +47,16 @@ export interface OffhostBackupConfig {
accessKey: string;
secretKey: string;
region: string;
/** The key new objects are written under (`BACKUP_ENCRYPTION_KEY`). */
encryptionKey: Buffer;
/**
* Keys older objects may still be under (`BACKUP_ENCRYPTION_PREVIOUS_KEYS`,
* comma-separated). Read only; nothing is written under them. This is the
* rotation path: the new key goes into `BACKUP_ENCRYPTION_KEY`, the old one
* here, until the bucket's lifecycle rule has retired every object written
* under it.
*/
previousEncryptionKeys: Buffer[];
}
export class OffhostBackupNotConfiguredError extends Error {
@@ -94,8 +109,8 @@ export function loadOffhostConfig(): OffhostBackupConfig | null {
// `BACKUP_RETENTION_DAYS` is deliberately absent from this config. It used
// to be parsed and clamped here and then read by nobody, which read as an
// enforcer the worker is not: retention belongs to the bucket's lifecycle
// rule, and no consumer can ever appear here because DeleteObject is kept
// out of the worker's grant on purpose (see the header). The variable stays
// rule, which the worker leaves to the bucket rather than deleting by age
// (see the header). The variable stays
// documented and on the compose whitelist because it is the number the
// operator sets that rule to.
return {
@@ -105,6 +120,11 @@ export function loadOffhostConfig(): OffhostBackupConfig | null {
secretKey,
region: envOr("BACKUP_S3_REGION", "auto"),
encryptionKey: decodeBackupKey(encRaw),
previousEncryptionKeys: (process.env.BACKUP_ENCRYPTION_PREVIOUS_KEYS ?? "")
.split(",")
.map((raw) => raw.trim())
.filter((raw) => raw.length > 0)
.map(decodeBackupKey),
};
}
@@ -115,6 +135,9 @@ export function loadOffhostConfig(): OffhostBackupConfig | null {
* 1: magic(4)="HLBK" || 0x01 || iv(12) || tag(16) || ciphertext(json)
* 2: magic(4)="HLBK" || 0x02 || iv(12) || tag(16) || ciphertext(gzip(json))
* 3: magic(4)="HLBK" || 0x03 || iv(12) || ciphertext(gzip(json)) || tag(16)
* 4: magic(4)="HLBK" || 0x04 || keyIdLen(1) || keyId || iv(12)
* || ciphertext(gzip(json)) || tag(16), with the object's own key
* (`<date>/user-<id>.json.enc`) as GCM associated data
*
* Version 1 encrypted the JSON directly; version 2 gzipped it first. Both put
* the tag in front of the ciphertext, and that is precisely what could not be
@@ -135,10 +158,108 @@ export function loadOffhostConfig(): OffhostBackupConfig | null {
const BACKUP_ENVELOPE_PLAIN = 0x01;
const BACKUP_ENVELOPE_GZIP = 0x02;
const BACKUP_ENVELOPE_STREAM = 0x03;
/**
* Version 4 adds two things version 3 did not have, and changes nothing else.
*
* A key id: the first twelve hex characters of the key's SHA-256, so a reader
* holding several keys (`BACKUP_ENCRYPTION_PREVIOUS_KEYS` during a rotation)
* knows which one opens an object instead of trying each. It is a
* fingerprint, not a name the operator has to keep in step.
*
* Associated data: the object's key in the bucket, which names the account
* and the night. Before this, an object copied to another account's key in
* the bucket, or to another date, decrypted as if it belonged there; now it
* does not open at all.
*/
const BACKUP_ENVELOPE_KEYED = 0x04;
const OFFHOST_AAD_PREFIX = "healthlog/offhost-backup/v4|";
const MAGIC = "HLBK";
/** magic(4) + version(1). Where the per-version body begins. */
const PREAMBLE_LENGTH = 5;
/** The off-host keys a reader may use: the current one, then the retired. */
export interface OffhostKeyRing {
active: Buffer;
previous: Buffer[];
}
/** The key ring of a loaded configuration. */
export function offhostKeyRing(cfg: OffhostBackupConfig): OffhostKeyRing {
return { active: cfg.encryptionKey, previous: cfg.previousEncryptionKeys };
}
function asKeyRing(key: Buffer | OffhostKeyRing): OffhostKeyRing {
return Buffer.isBuffer(key) ? { active: key, previous: [] } : key;
}
/** The id a version-4 object records for the key it was written under. */
export function offhostKeyId(key: Buffer): string {
return createHash("sha256").update(key).digest("hex").slice(0, 12);
}
function offhostAad(objectKey: string): Buffer {
return Buffer.from(`${OFFHOST_AAD_PREFIX}${objectKey}`, "utf8");
}
/**
* Open the gzip bytes of a version-4 object: find its key by id, verify the
* tag with the object key as associated data.
*/
function openKeyed(
buf: Buffer,
ring: OffhostKeyRing,
objectKey: string | undefined,
): Buffer {
if (objectKey === undefined) {
throw new Error(
"This backup object is bound to its key in the bucket; pass the object key to open it",
);
}
const idLen = buf[PREAMBLE_LENGTH];
const keyId = buf
.subarray(PREAMBLE_LENGTH + 1, PREAMBLE_LENGTH + 1 + idLen)
.toString("latin1");
const key = [ring.active, ...ring.previous].find(
(candidate) => offhostKeyId(candidate) === keyId,
);
if (!key) {
throw new Error(
`Backup object was written under off-host key ${keyId}, which is neither BACKUP_ENCRYPTION_KEY nor in BACKUP_ENCRYPTION_PREVIOUS_KEYS`,
);
}
return decryptRawStream(
buf.subarray(PREAMBLE_LENGTH + 1 + idLen),
key,
offhostAad(objectKey),
);
}
/**
* Try each key of the ring on an object with no key id (versions 1 to 3),
* current first. GCM refuses a wrong key outright, so the first key that
* opens it is the one it was written under.
*/
function withEachKey<T>(ring: OffhostKeyRing, open: (key: Buffer) => T): T {
let lastError: unknown = null;
for (const key of [ring.active, ...ring.previous]) {
try {
return open(key);
} catch (err) {
lastError = err;
}
}
throw lastError ?? new Error("No off-host key configured");
}
function isKnownVersion(version: number): boolean {
return (
version === BACKUP_ENVELOPE_PLAIN ||
version === BACKUP_ENVELOPE_GZIP ||
version === BACKUP_ENVELOPE_STREAM ||
version === BACKUP_ENVELOPE_KEYED
);
}
/**
* Write a whole JSON string as a version-2 object.
*
@@ -163,23 +284,36 @@ export function encryptBackup(plaintext: string, key: Buffer): Buffer {
return Buffer.concat([header, iv, tag, ct]);
}
export function decryptBackup(buf: Buffer, key: Buffer): string {
export function decryptBackup(
buf: Buffer,
keys: Buffer | OffhostKeyRing,
objectKey?: string,
): string {
const magic = buf.subarray(0, 4).toString("binary");
const version = buf[PREAMBLE_LENGTH - 1];
if (
magic !== MAGIC ||
(version !== BACKUP_ENVELOPE_PLAIN &&
version !== BACKUP_ENVELOPE_GZIP &&
version !== BACKUP_ENVELOPE_STREAM)
) {
if (magic !== MAGIC || !isKnownVersion(version)) {
throw new Error("Invalid backup envelope (bad magic or version)");
}
const ring = asKeyRing(keys);
if (version === BACKUP_ENVELOPE_KEYED) {
return gunzipSync(openKeyed(buf, ring, objectKey)).toString("utf8");
}
if (version === BACKUP_ENVELOPE_STREAM) {
// iv | ciphertext | tag, exactly what the streaming writer emits and what
// the shared reader verifies whole before it hands back a byte.
const plaintext = decryptRawStream(buf.subarray(PREAMBLE_LENGTH), key);
const plaintext = withEachKey(ring, (key) =>
decryptRawStream(buf.subarray(PREAMBLE_LENGTH), key),
);
return gunzipSync(plaintext).toString("utf8");
}
const plaintext = withEachKey(ring, (key) => openLeadingTag(buf, key));
return version === BACKUP_ENVELOPE_GZIP
? gunzipSync(plaintext).toString("utf8")
: plaintext.toString("utf8");
}
/** Versions 1 and 2: iv | tag | ciphertext. */
function openLeadingTag(buf: Buffer, key: Buffer): Buffer {
const iv = buf.subarray(PREAMBLE_LENGTH, PREAMBLE_LENGTH + IV_LENGTH);
const tag = buf.subarray(
PREAMBLE_LENGTH + IV_LENGTH,
@@ -188,10 +322,7 @@ export function decryptBackup(buf: Buffer, key: Buffer): string {
const ct = buf.subarray(PREAMBLE_LENGTH + IV_LENGTH + TAG_LENGTH);
const dec = createDecipheriv(ALGORITHM, key, iv);
dec.setAuthTag(tag);
const plaintext = Buffer.concat([dec.update(ct), dec.final()]);
return version === BACKUP_ENVELOPE_GZIP
? gunzipSync(plaintext).toString("utf8")
: plaintext.toString("utf8");
return Buffer.concat([dec.update(ct), dec.final()]);
}
/**
@@ -204,31 +335,24 @@ export function decryptBackup(buf: Buffer, key: Buffer): string {
*/
export function openBackupObject(
buf: Buffer,
key: Buffer,
keys: Buffer | OffhostKeyRing,
objectKey?: string,
): () => AsyncIterable<Buffer> {
const magic = buf.subarray(0, 4).toString("binary");
const version = buf[PREAMBLE_LENGTH - 1];
if (
magic !== MAGIC ||
(version !== BACKUP_ENVELOPE_PLAIN &&
version !== BACKUP_ENVELOPE_GZIP &&
version !== BACKUP_ENVELOPE_STREAM)
) {
if (magic !== MAGIC || !isKnownVersion(version)) {
throw new Error("Invalid backup envelope (bad magic or version)");
}
const ring = asKeyRing(keys);
let plaintext: Buffer;
if (version === BACKUP_ENVELOPE_STREAM) {
plaintext = decryptRawStream(buf.subarray(PREAMBLE_LENGTH), key);
} else {
const iv = buf.subarray(PREAMBLE_LENGTH, PREAMBLE_LENGTH + IV_LENGTH);
const tag = buf.subarray(
PREAMBLE_LENGTH + IV_LENGTH,
PREAMBLE_LENGTH + IV_LENGTH + TAG_LENGTH,
if (version === BACKUP_ENVELOPE_KEYED) {
plaintext = openKeyed(buf, ring, objectKey);
} else if (version === BACKUP_ENVELOPE_STREAM) {
plaintext = withEachKey(ring, (key) =>
decryptRawStream(buf.subarray(PREAMBLE_LENGTH), key),
);
const ct = buf.subarray(PREAMBLE_LENGTH + IV_LENGTH + TAG_LENGTH);
const dec = createDecipheriv(ALGORITHM, key, iv);
dec.setAuthTag(tag);
plaintext = Buffer.concat([dec.update(ct), dec.final()]);
} else {
plaintext = withEachKey(ring, (key) => openLeadingTag(buf, key));
}
if (version === BACKUP_ENVELOPE_PLAIN) {
return () => Readable.from([plaintext]);
@@ -245,6 +369,9 @@ export function openBackupObject(
const UPLOAD_PART_BYTES = 8 * 1024 * 1024;
const UPLOAD_CONCURRENCY = 2;
/** Pages of 1 000 keys one listing may walk: 200 000 objects. */
const MAX_LIST_PAGES = 200;
/**
* The largest object one multipart upload can carry: S3 and every compatible
* target cap a multipart upload at 10 000 parts.
@@ -303,12 +430,15 @@ export class OffhostBackupTooLargeError extends Error {
*/
function createEnvelopeStream(
key: Buffer,
objectKey: string,
limitBytes: number,
): { stream: Transform; bytes: () => number } {
const encryptor = createRawStreamEncryptor(key);
const encryptor = createRawStreamEncryptor(key, offhostAad(objectKey));
const keyId = Buffer.from(offhostKeyId(key), "latin1");
const header = Buffer.concat([
Buffer.from(MAGIC, "binary"),
Buffer.from([BACKUP_ENVELOPE_STREAM]),
Buffer.from([BACKUP_ENVELOPE_KEYED, keyId.byteLength]),
keyId,
encryptor.iv,
]);
let written = 0;
@@ -395,6 +525,7 @@ export async function uploadEncryptedBackup(
const gzip = createGzip();
const { stream: envelope, bytes } = createEnvelopeStream(
encryptionKey,
objectKey,
limitBytes,
);
@@ -487,6 +618,24 @@ export interface S3Like {
prefix: string,
): Promise<Array<{ key: string; lastModified?: Date }>>;
deleteObject(key: string): Promise<void>;
/**
* The bucket's lifecycle rule, as far as the credential may read it.
* Optional so a test double that has no opinion leaves it out.
*/
getLifecycle?(): Promise<OffhostLifecycle>;
}
/**
* Whether the bucket retires old copies by itself.
*
* `configured` names the shortest expiry of an enabled rule; `missing` means
* the bucket answered that it has no rule; `unknown` means it would not say
* (a credential without `s3:GetLifecycleConfiguration`, a target that does not
* implement the call, a network failure) and says nothing either way.
*/
export interface OffhostLifecycle {
state: "configured" | "missing" | "unknown";
expirationDays: number | null;
}
export async function getS3Client(cfg: OffhostBackupConfig): Promise<S3Like> {
@@ -568,22 +717,109 @@ export async function getS3Client(cfg: OffhostBackupConfig): Promise<S3Like> {
}
},
listObjects: async (prefix) => {
const out = await client.send(
new mod.ListObjectsV2Command({ Bucket: cfg.bucket, Prefix: prefix }),
// Every page, not the first. A bucket answers at most 1 000 keys per
// call, in key order, so the first page of a bucket with a few accounts
// and a month of retention holds only the OLDEST dates: the restore
// drill read an old object and called the chain stale, and a purge
// would have missed every newer copy. Bounded, so a bucket that holds
// far more than this job put there cannot keep the loop going forever.
const found: Array<{ key: string; lastModified?: Date }> = [];
let token: string | undefined;
for (let page = 0; page < MAX_LIST_PAGES; page++) {
const out = await client.send(
new mod.ListObjectsV2Command({
Bucket: cfg.bucket,
Prefix: prefix,
ContinuationToken: token,
}),
);
for (const c of out.Contents ?? []) {
found.push({ key: c.Key ?? "", lastModified: c.LastModified });
}
if (!out.IsTruncated || !out.NextContinuationToken) return found;
token = out.NextContinuationToken;
}
throw new Error(
`Bucket listing under "${prefix}" did not end within ${MAX_LIST_PAGES} pages`,
);
return (out.Contents ?? []).map((c) => ({
key: c.Key ?? "",
lastModified: c.LastModified,
}));
},
deleteObject: async (key) => {
await client.send(
new mod.DeleteObjectCommand({ Bucket: cfg.bucket, Key: key }),
);
},
getLifecycle: async () => {
try {
const out = await client.send(
new mod.GetBucketLifecycleConfigurationCommand({
Bucket: cfg.bucket,
}),
{ abortSignal: AbortSignal.timeout(5_000) },
);
const days = (out.Rules ?? [])
.filter((rule) => rule.Status === "Enabled")
.map((rule) => rule.Expiration?.Days)
.filter((d): d is number => typeof d === "number" && d > 0);
return days.length > 0
? { state: "configured", expirationDays: Math.min(...days) }
: { state: "missing", expirationDays: null };
} catch (err) {
const name = (err as { name?: string; Code?: string }).name;
const code = (err as { Code?: string }).Code;
if (
name === "NoSuchLifecycleConfiguration" ||
code === "NoSuchLifecycleConfiguration"
) {
return { state: "missing", expirationDays: null };
}
return { state: "unknown", expirationDays: null };
}
},
};
}
const LIFECYCLE_CACHE_MS = 10 * 60 * 1000;
let lifecycleCache: { at: number; value: OffhostLifecycle } | null = null;
/**
* The bucket's lifecycle rule for the admin card, cached for ten minutes so a
* page render does not become a bucket call. Never throws.
*/
export async function probeOffhostLifecycle(
s3Override?: S3Like,
now: number = Date.now(),
): Promise<OffhostLifecycle> {
if (
!s3Override &&
lifecycleCache &&
now - lifecycleCache.at < LIFECYCLE_CACHE_MS
) {
return lifecycleCache.value;
}
const cfg = loadOffhostConfigSafe();
if (!cfg) return { state: "unknown", expirationDays: null };
let value: OffhostLifecycle;
try {
const s3 = s3Override ?? (await getS3Client(cfg));
value = s3.getLifecycle
? await s3.getLifecycle()
: { state: "unknown", expirationDays: null };
} catch {
value = { state: "unknown", expirationDays: null };
}
if (!s3Override) lifecycleCache = { at: now, value };
return value;
}
/** `loadOffhostConfig`, answering null for a malformed key instead of throwing. */
function loadOffhostConfigSafe(): OffhostBackupConfig | null {
try {
return loadOffhostConfig();
} catch {
return null;
}
}
interface BackupRunReport {
config: { endpoint: string; bucket: string; region: string };
uploaded: number;
@@ -622,6 +858,30 @@ export interface RunOffhostBackupOptions extends UploadBackupOptions {
shouldStop?: () => boolean;
}
/**
* Note, per key id, that an object needing it went into the bucket now. Never
* fails the account: the object is in the bucket whatever this row says.
*/
async function recordOffhostKeyUse(
prisma: PrismaClient,
keyIds: readonly string[],
at: Date,
): Promise<void> {
for (const keyId of keyIds) {
try {
await prisma.offhostBackupKeyUse.upsert({
where: { keyId },
create: { keyId, firstWrittenAt: at, lastWrittenAt: at },
update: { lastWrittenAt: at },
});
} catch (err) {
getEvent()?.addWarning(
`offhost-backup key-use write failed: ${(err as Error).message?.slice(0, 200)}`,
);
}
}
}
export async function runOffhostBackup(
prisma: PrismaClient,
s3Override?: S3Like,
@@ -669,24 +929,54 @@ export async function runOffhostBackup(
break;
}
let objectBytes: number | null = null;
const objectKey = `${dateKey}/user-${user.id}.json.enc`;
const accountStartedAt = new Date();
// The key ids the object's content needs, read as it is written: the
// admin encryption view uses them to say how long a retired key is still
// needed for what is in the bucket.
const keyScanner = createBackupKeyIdTextScanner();
try {
objectBytes = await uploadEncryptedBackup(
s3,
`${dateKey}/user-${user.id}.json.enc`,
objectKey,
cfg.encryptionKey,
// The same writer the weekly in-database pass uses. The payload
// builder was always shared; everything after it was not, which is why
// this job kept dying on a record the weekly one had learned to
// survive.
(write) =>
streamFullBackupJson(prisma, user.id, write, {
purpose: "disaster-recovery",
exportedAt: now,
}),
streamFullBackupJson(
prisma,
user.id,
async (chunk) => {
keyScanner.feed(chunk);
await write(chunk);
},
{
purpose: "disaster-recovery",
exportedAt: now,
},
),
options,
);
// The account was deleted or wiped while its copy was being written.
// The purge that request started may already have run, so the copy
// this run just put there is removed here rather than left behind.
const purgedMeanwhile =
(await prisma.offhostPurgeRequest.count({
where: {
subjectId: user.id,
requestedAt: { gte: accountStartedAt },
},
})) > 0 || (await prisma.user.count({ where: { id: user.id } })) === 0;
if (purgedMeanwhile) {
await s3.deleteObject(objectKey);
objectBytes = null;
continue;
}
largestObjectBytes = Math.max(largestObjectBytes, objectBytes);
uploaded++;
await recordOffhostKeyUse(prisma, keyScanner.keyIds(), new Date());
} catch (err) {
failed++;
if (err instanceof OffhostBackupTooLargeError) oversized++;
+260
View File
@@ -0,0 +1,260 @@
/**
* Off-host copies leave the bucket with the account.
*
* The nightly job writes `<YYYY-MM-DD>/user-<id>.json.enc` for every account,
* and until v1.39.3 nothing ever removed one: deleting an account, or wiping
* its data, took the rows in this database and left every nightly copy of the
* record in the operator's bucket for as long as its lifecycle rule kept them,
* or forever where no rule was set. A confirmation that says "permanently
* delete" has to mean the off-host copies too.
*
* How. The deletion writes an `OffhostPurgeRequest` in its own transaction, so
* the request exists exactly when the deletion committed. This job then lists
* the bucket once, deletes every object of every pending account, and removes
* the request when nothing of the account is left. A request the bucket
* refuses (a credential without DeleteObject, an outage) stays, counts its
* attempts and keeps the bucket's answer, which the admin off-host card shows;
* the job runs again every night and after every nightly upload, so a late
* copy the upload wrote for an account deleted while it ran is caught too.
*
* The admin wipe of every account writes no request, on purpose: it is the
* operator clearing their own host, the bucket is theirs, and its copies are
* the documented way back from a wipe pressed by mistake. The health-check
* probes and anything else in the bucket that is not an account object is
* never touched.
*/
import type { Job } from "pg-boss";
import type { Prisma, PrismaClient } from "@/generated/prisma/client";
import { auditLog } from "@/lib/auth/audit";
import { getGlobalBoss } from "@/lib/jobs/boss-instance";
import { jobDone, jobFailed, type JobOutcome } from "@/lib/jobs/job-outcome";
import {
getS3Client,
loadOffhostConfig,
offhostBackupConfigured,
type S3Like,
} from "@/lib/jobs/offhost-backup";
import { reportWorkerError } from "@/lib/jobs/report-worker-error";
import { getWorkerPrisma } from "@/lib/jobs/reminder/shared";
import { withBackgroundEvent } from "@/lib/logging/background";
export const OFFHOST_PURGE_QUEUE = "offhost-backup-purge";
/** 03:40, after the nightly upload (02:30) has had its hour. */
export const OFFHOST_PURGE_CRON = "40 3 * * *";
export type OffhostPurgeReason =
"account_deleted" | "data_wiped" | "managed_profile_deleted";
/** An account object: `<date>/user-<id>.json.enc`. */
const ACCOUNT_OBJECT = /^\d{4}-\d{2}-\d{2}\/user-(.+)\.json\.enc$/;
/** Requests one run works through; the next run takes the rest. */
const REQUESTS_PER_RUN = 200;
/** Deletes one run may issue, so a huge bucket cannot run the job forever. */
const MAX_DELETES_PER_RUN = 50_000;
type Db = Pick<Prisma.TransactionClient, "offhostPurgeRequest">;
/**
* Record that `subjectId`'s off-host copies have to go. Call it with the
* transaction that deletes the account or wipes its data, so the request and
* the deletion commit together. A host without off-host backup configured has
* nothing in any bucket this could reach, and records nothing.
*/
export async function requestOffhostPurge(
db: Db,
subjectId: string,
reason: OffhostPurgeReason,
): Promise<boolean> {
if (!offhostBackupConfigured()) return false;
await db.offhostPurgeRequest.create({
data: { subjectId, reason },
select: { id: true },
});
return true;
}
/**
* Wake the purge job now rather than at its nightly slot. Best effort: the
* request row is what guarantees the purge, and the cron picks it up if the
* queue is not reachable from here.
*/
export async function kickOffhostPurge(): Promise<void> {
try {
await getGlobalBoss()?.send(
OFFHOST_PURGE_QUEUE,
{},
{
singletonKey: OFFHOST_PURGE_QUEUE,
retryLimit: 3,
retryDelay: 300,
},
);
} catch {
// The cron is the backstop.
}
}
export interface OffhostPurgeReport {
pending: number;
completed: number;
failed: number;
objectsDeleted: number;
}
function trimError(err: unknown): string {
return (err instanceof Error ? err.message : String(err)).slice(0, 300);
}
/**
* Work through the pending requests: list the bucket once, delete every
* object of every pending subject, remove each request whose objects are all
* gone. Answers what it did.
*/
export async function processOffhostPurges(
prisma: PrismaClient,
s3Override?: S3Like,
): Promise<OffhostPurgeReport> {
const requests = await prisma.offhostPurgeRequest.findMany({
orderBy: { requestedAt: "asc" },
take: REQUESTS_PER_RUN,
});
const report: OffhostPurgeReport = {
pending: requests.length,
completed: 0,
failed: 0,
objectsDeleted: 0,
};
if (requests.length === 0) return report;
const cfg = loadOffhostConfig();
if (!cfg) {
// Configuration removed after the requests were written: leave them, an
// operator who puts it back gets the purge.
return report;
}
const now = new Date();
const markFailed = async (ids: string[], err: unknown) => {
await prisma.offhostPurgeRequest.updateMany({
where: { id: { in: ids } },
data: {
attempts: { increment: 1 },
lastAttemptAt: now,
lastFailure: trimError(err),
},
});
report.failed += ids.length;
};
let objects: Array<{ key: string }>;
let s3: S3Like;
try {
s3 = s3Override ?? (await getS3Client(cfg));
objects = await s3.listObjects("");
} catch (err) {
await markFailed(
requests.map((r) => r.id),
err,
);
return report;
}
const subjects = new Set(requests.map((r) => r.subjectId));
const bySubject = new Map<string, string[]>();
for (const { key } of objects) {
const match = ACCOUNT_OBJECT.exec(key);
if (!match) continue;
const subject = match[1];
if (!subjects.has(subject)) continue;
const list = bySubject.get(subject) ?? [];
list.push(key);
bySubject.set(subject, list);
}
// One subject at a time, so a refusal part-way fails only the subjects it
// reached, and a request is removed only when its own objects are gone.
const failedSubjects = new Map<string, unknown>();
let deletes = 0;
for (const [subject, keys] of bySubject) {
for (const key of keys) {
if (deletes >= MAX_DELETES_PER_RUN) {
failedSubjects.set(
subject,
new Error("Deletion budget for one run reached; continues next run"),
);
break;
}
try {
await s3.deleteObject(key);
deletes++;
report.objectsDeleted++;
} catch (err) {
failedSubjects.set(subject, err);
break;
}
}
}
for (const request of requests) {
const failure = failedSubjects.get(request.subjectId);
if (failure) {
await markFailed([request.id], failure);
continue;
}
const deleted = bySubject.get(request.subjectId)?.length ?? 0;
await prisma.offhostPurgeRequest.delete({ where: { id: request.id } });
report.completed++;
// The account a deleted subject named is gone, and so is its audit
// history; the receipt names the reason and the count, not the account.
// A wiped account still exists and keeps the receipt as its own.
const accountRemains =
request.reason === "data_wiped" &&
(await prisma.user.count({ where: { id: request.subjectId } })) > 0;
await auditLog("offhost.backup.purged", {
userId: accountRemains ? request.subjectId : null,
actorUserId: null,
details: {
reason: request.reason,
objectsDeleted: deleted,
attempts: request.attempts + 1,
},
});
}
return report;
}
export async function handleOffhostPurge(
jobs: Job<object>[],
): Promise<JobOutcome> {
void jobs;
return withBackgroundEvent("job.offhost_purge", async (evt) => {
try {
const report = await processOffhostPurges(getWorkerPrisma());
evt.addMeta("offhost_purge_pending", report.pending);
evt.addMeta("offhost_purge_completed", report.completed);
evt.addMeta("offhost_purge_failed", report.failed);
evt.addMeta("offhost_purge_objects_deleted", report.objectsDeleted);
if (report.failed > 0) {
await reportWorkerError(
OFFHOST_PURGE_QUEUE,
new Error(
`${report.failed} off-host deletion request(s) could not be completed; they stay pending and are retried nightly. Check that the bucket credential allows DeleteObject.`,
),
);
}
return jobDone({
offhost_purge_pending: report.pending,
offhost_purge_completed: report.completed,
offhost_purge_failed: report.failed,
offhost_purge_objects_deleted: report.objectsDeleted,
});
} catch (err) {
evt.addWarning(`offhost-purge failed: ${err}`);
await reportWorkerError(OFFHOST_PURGE_QUEUE, err);
return jobFailed("offhost purge failed", err);
}
});
}
@@ -67,6 +67,11 @@ import {
enqueueBootTimeIntakeSlotDedup,
type IntakeSlotDedupPayload,
} from "@/lib/medications/intake-slot-dedup";
import {
handleOffhostPurge,
OFFHOST_PURGE_CRON,
OFFHOST_PURGE_QUEUE,
} from "@/lib/jobs/offhost-purge";
import {
handleRestoreDrill,
RESTORE_DRILL_CRON,
@@ -353,6 +358,10 @@ const allQueues = [
STEP_UP_ELEVATION_CLEANUP_QUEUE,
OFFHOST_BACKUP_QUEUE,
RESTORE_DRILL_QUEUE,
// v1.39.3 — deleting an account or wiping its data takes its off-host
// copies out of the bucket. Without this entry the requests would be
// written and never worked.
OFFHOST_PURGE_QUEUE,
// v1.39.1 — an admin's restore of a stored backup, run off the request, and
// the boot sweep that re-queues one a stopped worker left running. Without
// this entry every restore request would queue a job nobody works.
@@ -505,6 +514,9 @@ const schedules: ScheduleEntry[] = [
],
[OFFHOST_BACKUP_QUEUE, OFFHOST_BACKUP_CRON],
[RESTORE_DRILL_QUEUE, RESTORE_DRILL_CRON],
// Nightly backstop for the purge the deletion kicks straight away: a
// request the bucket refused, or one written while the queue was down.
[OFFHOST_PURGE_QUEUE, OFFHOST_PURGE_CRON, cronIsTheRetry],
[HOST_METRIC_QUEUE, HOST_METRIC_CRON, cronIsTheRetry],
[FEEDBACK_AGGREGATOR_QUEUE, FEEDBACK_AGGREGATOR_CRON, cronIsTheRetry],
// v1.4.37 — hourly geo backfill. The helper is idempotent + capped
@@ -747,6 +759,12 @@ export async function registerMaintenanceQueues(
{ localConcurrency: 1 },
handleRestoreDrill,
);
await createAndWork(
boss,
OFFHOST_PURGE_QUEUE,
{ localConcurrency: 1 },
handleOffhostPurge,
);
// One restore at a time per worker: each holds a transaction over a whole
// account and a batch of readings in memory, and two at once on a small host
// is how a restore that fits alone would run out of heap.
+29 -1
View File
@@ -23,8 +23,14 @@
*/
import type { Job } from "pg-boss";
import { BackupJsonError, scanBackupJson } from "@/lib/export/backup-json-scan";
import {
assessBackupKeys,
BackupKeyIdCollector,
describeBackupKeyProblem,
} from "@/lib/export/backup-key-ids";
import {
openBackupObject,
offhostKeyRing,
getS3Client,
loadOffhostConfig,
OffhostBackupNotConfiguredError,
@@ -54,6 +60,8 @@ export interface RestoreDrillReport {
stale: boolean;
ciphertextBytes: number;
plaintextBytes: number;
/** Application key ids the object's inner ciphertext needs, each proven to open. */
innerKeyIds: string[];
recordCounts: {
measurements: number;
medications: number;
@@ -107,7 +115,7 @@ export async function runRestoreDrill(
// Read as a stream, never as one string: the JSON of a large record is
// longer than any string V8 can hold (#1031). The bulk tables are counted,
// not kept.
const source = openBackupObject(ciphertext, cfg.encryptionKey);
const source = openBackupObject(ciphertext, offhostKeyRing(cfg), objectKey);
let plaintextBytes = 0;
async function* counted() {
for await (const chunk of source()) {
@@ -115,10 +123,17 @@ export async function runRestoreDrill(
yield chunk;
}
}
// The envelope opening proves the off-host key. It proves nothing about
// the application keys the ciphertext INSIDE was written under, and that
// is the half a key rotation breaks: a restore of this object writes those
// values back verbatim. So every inner key id is collected, and one value
// per key is actually decrypted.
const keys = new BackupKeyIdCollector();
let scanned;
try {
scanned = await scanBackupJson(counted(), {
streamKeys: new Set(["measurements", "intakeEvents", "moodEntries"]),
onElement: (key, element) => keys.visit(element, key),
});
} catch (err) {
if (err instanceof BackupJsonError) {
@@ -140,6 +155,17 @@ export async function runRestoreDrill(
);
}
keys.visit(payload);
const keyVerdict = assessBackupKeys(keys, {
ignoreSections: new Set(["appSettings"]),
});
const keyProblem = describeBackupKeyProblem(keyVerdict);
if (keyProblem) {
throw new Error(
`Restore drill: backup object "${objectKey}" opens, but its content could not be restored on this server. ${keyProblem}`,
);
}
const ageDays = Math.floor(
(now.getTime() - Date.parse(`${dateKey}T00:00:00Z`)) /
(24 * 60 * 60 * 1000),
@@ -152,6 +178,7 @@ export async function runRestoreDrill(
stale: ageDays > MAX_BACKUP_AGE_DAYS,
ciphertextBytes: ciphertext.length,
plaintextBytes,
innerKeyIds: keyVerdict.keyIds,
recordCounts: {
measurements: streamedCounts.measurements,
medications: countArray(payload.medications),
@@ -193,6 +220,7 @@ export async function handleRestoreDrill(
report.recordCounts.moodEntries,
);
evt.addMeta("restore_drill_stale", report.stale);
evt.addMeta("restore_drill_inner_key_ids", report.innerKeyIds.join(","));
if (report.stale) {
await reportWorkerError(
RESTORE_DRILL_QUEUE,
+1
View File
@@ -101,6 +101,7 @@ const subsystemSurface = {
"step-up-elevation-cleanup": { audience: "system" },
"data-backup-offhost": { audience: "system" },
"data-restore-drill": { audience: "system" },
"offhost-backup-purge": { audience: "system" },
"backup-restore": { audience: "system" },
"host-metric-sample": { audience: "system" },
"feedback-aggregator": { audience: "system" },
+6
View File
@@ -1,3 +1,7 @@
import {
kickOffhostPurge,
requestOffhostPurge,
} from "@/lib/jobs/offhost-purge";
import { prisma } from "@/lib/db";
import { auditLog } from "@/lib/auth/audit";
import type { Prisma } from "@/generated/prisma/client";
@@ -209,9 +213,11 @@ export async function deleteManagedProfile(input: {
details: { profileId: profile.id },
client: tx,
});
await requestOffhostPurge(tx, profile.id, "managed_profile_deleted");
await tx.user.delete({ where: { id: profile.id } });
});
});
await kickOffhostPurge();
}
/**
@@ -1,5 +1,7 @@
import { describe, expect, it } from "vitest";
import { readFileSync, unlinkSync } from "node:fs";
import { mkdtempSync, readdirSync, readFileSync, unlinkSync } from "node:fs";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { createHash } from "node:crypto";
import { deflateRawSync } from "node:zlib";
@@ -112,6 +114,29 @@ describe("streamMultipartToDisk", () => {
).rejects.toThrow(/size cap/);
});
// What was written before the refusal is the person's export in plain
// text; it must not stay behind in the temp directory.
it("removes the partial file when the upload is refused part-way", async () => {
const dir = mkdtempSync(join(tmpdir(), "healthlog-stream-"));
const body = buildMultipartBody("export.zip", Buffer.alloc(4096, 0x41));
const chunked = new ReadableStream<Uint8Array>({
start(controller) {
for (let i = 0; i < body.length; i += 256) {
controller.enqueue(new Uint8Array(body.subarray(i, i + 256)));
}
controller.close();
},
});
await expect(
streamMultipartToDisk(
chunked,
`multipart/form-data; boundary=${BOUNDARY}`,
{ maxBytes: 1000, fieldName: "file", tmpDir: dir },
),
).rejects.toThrow(/size cap/);
expect(readdirSync(dir)).toEqual([]);
});
it("throws when the named field is missing", async () => {
const body = buildMultipartBody("export.zip", Buffer.from("x"), {
something: "else",
+33 -13
View File
@@ -17,6 +17,7 @@
*/
import { createHash, randomUUID } from "node:crypto";
import { createWriteStream } from "node:fs";
import { rm } from "node:fs/promises";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { Writable } from "node:stream";
@@ -293,22 +294,41 @@ export async function streamMultipartToDisk(
}
};
while (!streamDone) {
const { value, done } = await reader.read();
if (done) {
streamDone = true;
break;
try {
while (!streamDone) {
const { value, done } = await reader.read();
if (done) {
streamDone = true;
break;
}
buffer = Buffer.concat([buffer, Buffer.from(value)]);
await processBuffer();
}
buffer = Buffer.concat([buffer, Buffer.from(value)]);
await processBuffer();
}
await processBuffer();
await closeFileSink();
await closeFileSink();
if (bytesWritten === 0) {
throw new Error(
`Multipart body did not include a '${opts.fieldName}' file field`,
);
if (bytesWritten === 0) {
throw new Error(
`Multipart body did not include a '${opts.fieldName}' file field`,
);
}
} catch (err) {
// Whatever was written so far is the user's plaintext health export,
// and it must not outlive a refused upload in the shared temp
// directory: the size cap, a malformed body and a dropped connection
// all end here. The reader is cancelled so the rest of the body is not
// read for nothing.
await reader.cancel().catch(() => {});
const sink = fileSink as Writable | null;
if (sink && !sink.closed) {
fileSinkOpen = false;
await new Promise<void>((resolve) => {
sink.once("close", () => resolve());
sink.destroy();
});
}
await rm(filePath, { force: true }).catch(() => {});
throw err;
}
return {
+3 -1
View File
@@ -69,6 +69,8 @@ export const ERROR_CODE_CATALOGUE: Readonly<Record<string, readonly string[]>> =
"auth.token.invalid",
],
backup: [
"backup.foreign_reference",
"backup.key.missing",
"backup.payload.undecryptable",
"backup.restore.active",
"backup.section.missing",
@@ -185,7 +187,7 @@ export const ERROR_CODE_CATALOGUE: Readonly<Record<string, readonly string[]>> =
"illness.episode.parent-not-found",
"illness.insights.invalid",
],
import: ["import.write_failed"],
import: ["import.apple_health.busy", "import.write_failed"],
ingest: ["ingest.rate_limited"],
insights: [
"insights.generate.budgetExceeded",
+3 -2
View File
@@ -86,15 +86,16 @@ const restoreFailure = z
.object({
code: z.string().meta({
description:
"Stable reason: `backup_not_found`, `backup_changed`, `backup.payload.undecryptable`, `schema_invalid`, `incompatible_schema_version`, `owner_mismatch`, `owner_not_found`, `backup.section.missing`, `document_ciphertext_missing`, `time_budget`, `transaction_failed`, `interrupted`, `not_started`, `enqueue_failed`, `failed_after_commit`, `unexpected`. Every code but `failed_after_commit` means the account was not changed.",
"Stable reason: `backup_not_found`, `backup_changed`, `backup.payload.undecryptable`, `schema_invalid`, `incompatible_schema_version`, `owner_mismatch`, `owner_not_found`, `backup.section.missing`, `document_ciphertext_missing`, `backup.key.missing`, `backup.foreign_reference`, `time_budget`, `transaction_failed`, `interrupted`, `not_started`, `enqueue_failed`, `failed_after_commit`, `unexpected`. Every code but `failed_after_commit` means the account was not changed.",
}),
message: z.string(),
sections: z.array(z.string()).optional(),
keyIds: z.array(z.string()).optional(),
})
.meta({
id: "AdminBackupRestoreFailure",
description:
"Why the restore did not happen, in a sentence an operator can act on. `sections` names the missing sections for `backup.section.missing`.",
"Why the restore did not happen, in a sentence an operator can act on. `sections` names the missing sections for `backup.section.missing`; `keyIds` names the encryption key ids the file needs and this server cannot open it with, for `backup.key.missing`.",
});
const restoreJob = z
+6 -1
View File
@@ -324,6 +324,11 @@ export const importPaths: NonNullable<ZodOpenApiObject["paths"]> = {
content: { "application/json": { schema: errorEnvelope } },
},
...stdResponses,
"409": {
description:
"An import for this account is already queued or running, and these bytes are not the ones it carries (the same bytes resolve to that job as a 202). `meta.errorCode` is `import.apple_health.busy` and `meta.jobId` names the running job; poll it, then upload again. The upload was discarded.",
content: { "application/json": { schema: errorEnvelope } },
},
"422": {
description:
"The multipart body could not be streamed to disk — no `file` part, a malformed boundary, or the size cap tripped mid-stream. The message names the failure.",
@@ -335,7 +340,7 @@ export const importPaths: NonNullable<ZodOpenApiObject["paths"]> = {
},
"503": {
description:
"No background worker is bound to this process, so the job cannot be queued. The upload was staged and is discarded; retry once a worker is running.",
"No background worker is bound to this process, or the queue did not accept the job, so nothing will run it. The staged upload is discarded (and a created job is marked failed); retry once a worker is running.",
content: { "application/json": { schema: errorEnvelope } },
},
},
+19
View File
@@ -58,6 +58,25 @@ export interface OffhostBackupOverview {
periodHours: number;
/** One row per account. Empty when off-host backup is not configured. */
rows: OffhostAccountRow[];
/**
* Deleted or wiped accounts whose copies are still in the bucket. A count
* above zero with an error means the bucket refused the deletion (most
* often a credential without DeleteObject); the job retries nightly.
*/
pendingDeletions: {
count: number;
oldestRequestedAt: string | null;
lastFailure: string | null;
};
/**
* Whether the bucket retires old copies by itself. Read from the bucket
* (`GetBucketLifecycleConfiguration`), cached for ten minutes; `unknown`
* when the credential may not read it.
*/
lifecycle: {
state: "configured" | "missing" | "unknown";
expirationDays: number | null;
};
}
export interface BackupsList {