refactor: move attachment upload and download logic to a separate js.
This commit is contained in:
qaz741wsd856
parent
58847e9788
commit
034e6284a2
2 changed files with 463 additions and 445 deletions
446
src/attachments.js
Normal file
446
src/attachments.js
Normal file
|
|
@ -0,0 +1,446 @@
|
|||
/**
|
||||
* Attachment upload/download fast-path for Warden Worker (JS)
|
||||
*
|
||||
* This module implements:
|
||||
* - Attachment upload logic (zero-copy streaming to R2)
|
||||
* - Attachment download logic (zero-copy streaming from R2)
|
||||
* - JWT validation for attachment tokens (HMAC-SHA256) using Web Crypto API
|
||||
*
|
||||
* Route matching and URL parsing should be handled by `src/entry.js`.
|
||||
*/
|
||||
|
||||
// JWT validation using Web Crypto API (no external dependencies)
|
||||
async function verifyJwt(token, secret) {
|
||||
const encoder = new TextEncoder();
|
||||
const parts = token.split(".");
|
||||
if (parts.length !== 3) {
|
||||
throw new Error("Invalid token format");
|
||||
}
|
||||
|
||||
const [headerB64, payloadB64, signatureB64] = parts;
|
||||
|
||||
// Import the secret key for HMAC-SHA256
|
||||
const key = await crypto.subtle.importKey(
|
||||
"raw",
|
||||
encoder.encode(secret),
|
||||
{ name: "HMAC", hash: "SHA-256" },
|
||||
false,
|
||||
["verify"]
|
||||
);
|
||||
|
||||
// Decode the signature (base64url to Uint8Array)
|
||||
const signature = base64UrlDecode(signatureB64);
|
||||
|
||||
// Verify the signature
|
||||
const data = encoder.encode(`${headerB64}.${payloadB64}`);
|
||||
const valid = await crypto.subtle.verify("HMAC", key, signature, data);
|
||||
|
||||
if (!valid) {
|
||||
throw new Error("Invalid token signature");
|
||||
}
|
||||
|
||||
// Decode and parse the payload
|
||||
const payload = JSON.parse(
|
||||
new TextDecoder().decode(base64UrlDecode(payloadB64))
|
||||
);
|
||||
|
||||
// Check expiration
|
||||
if (payload.exp && payload.exp < Math.floor(Date.now() / 1000)) {
|
||||
throw new Error("Token expired");
|
||||
}
|
||||
|
||||
return payload;
|
||||
}
|
||||
|
||||
export function base64UrlDecode(str) {
|
||||
// Convert base64url to base64
|
||||
let base64 = str.replace(/-/g, "+").replace(/_/g, "/");
|
||||
// Add padding if needed
|
||||
while (base64.length % 4) {
|
||||
base64 += "=";
|
||||
}
|
||||
const binary = atob(base64);
|
||||
const bytes = new Uint8Array(binary.length);
|
||||
for (let i = 0; i < binary.length; i++) {
|
||||
bytes[i] = binary.charCodeAt(i);
|
||||
}
|
||||
return bytes;
|
||||
}
|
||||
|
||||
// Generate ISO timestamp string
|
||||
function nowString() {
|
||||
return new Date().toISOString();
|
||||
}
|
||||
|
||||
// Helper to get env var with fallback
|
||||
function getEnvVar(env, name, defaultValue = null) {
|
||||
try {
|
||||
const value = env[name];
|
||||
if (value && typeof value.toString === "function") {
|
||||
return value.toString();
|
||||
}
|
||||
return value || defaultValue;
|
||||
} catch {
|
||||
return defaultValue;
|
||||
}
|
||||
}
|
||||
|
||||
// Get attachment size limits from env
|
||||
function getAttachmentMaxBytes(env) {
|
||||
const value = getEnvVar(env, "ATTACHMENT_MAX_BYTES");
|
||||
if (!value) return null;
|
||||
const parsed = parseInt(value, 10);
|
||||
return isNaN(parsed) ? null : parsed;
|
||||
}
|
||||
|
||||
function getTotalLimitBytes(env) {
|
||||
const value = getEnvVar(env, "ATTACHMENT_TOTAL_LIMIT_KB");
|
||||
if (!value) return null;
|
||||
const kb = parseInt(value, 10);
|
||||
if (isNaN(kb)) return null;
|
||||
return kb * 1024;
|
||||
}
|
||||
|
||||
// Get user's current attachment usage
|
||||
async function getUserAttachmentUsage(db, userId, excludeAttachmentId) {
|
||||
const query = excludeAttachmentId
|
||||
? `SELECT COALESCE(SUM(file_size), 0) as total FROM (
|
||||
SELECT a.file_size as file_size
|
||||
FROM attachments a
|
||||
JOIN ciphers c ON c.id = a.cipher_id
|
||||
WHERE c.user_id = ?1 AND a.id != ?2
|
||||
UNION ALL
|
||||
SELECT p.file_size as file_size
|
||||
FROM attachments_pending p
|
||||
JOIN ciphers c2 ON c2.id = p.cipher_id
|
||||
WHERE c2.user_id = ?1 AND p.id != ?2
|
||||
) AS files`
|
||||
: `SELECT COALESCE(SUM(file_size), 0) as total FROM (
|
||||
SELECT a.file_size as file_size
|
||||
FROM attachments a
|
||||
JOIN ciphers c ON c.id = a.cipher_id
|
||||
WHERE c.user_id = ?1
|
||||
UNION ALL
|
||||
SELECT p.file_size as file_size
|
||||
FROM attachments_pending p
|
||||
JOIN ciphers c2 ON c2.id = p.cipher_id
|
||||
WHERE c2.user_id = ?1
|
||||
) AS files`;
|
||||
|
||||
const bindings = excludeAttachmentId ? [userId, excludeAttachmentId] : [userId];
|
||||
|
||||
const result = await db.prepare(query).bind(...bindings).first();
|
||||
return result?.total || 0;
|
||||
}
|
||||
|
||||
// Enforce attachment size limits
|
||||
async function enforceLimits(db, env, userId, newSize, excludeAttachmentId) {
|
||||
if (newSize < 0) {
|
||||
throw new Error("Attachment size cannot be negative");
|
||||
}
|
||||
|
||||
const maxBytes = getAttachmentMaxBytes(env);
|
||||
if (maxBytes !== null && newSize > maxBytes) {
|
||||
throw new Error("Attachment size exceeds limit");
|
||||
}
|
||||
|
||||
const limitBytes = getTotalLimitBytes(env);
|
||||
if (limitBytes !== null) {
|
||||
const used = await getUserAttachmentUsage(db, userId, excludeAttachmentId);
|
||||
const newTotal = used + newSize;
|
||||
if (newTotal > limitBytes) {
|
||||
throw new Error("Attachment storage limit reached");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Handle azure-upload with zero-copy streaming
|
||||
export async function handleAzureUpload(request, env, cipherId, attachmentId, token) {
|
||||
// Get R2 bucket
|
||||
const bucket = env.ATTACHMENTS_BUCKET;
|
||||
if (!bucket) {
|
||||
return new Response(JSON.stringify({ error: "Attachments are not enabled" }), {
|
||||
status: 400,
|
||||
headers: { "Content-Type": "application/json" },
|
||||
});
|
||||
}
|
||||
|
||||
// Get D1 database
|
||||
const db = env.vault1;
|
||||
if (!db) {
|
||||
return new Response(JSON.stringify({ error: "Database not available" }), {
|
||||
status: 500,
|
||||
headers: { "Content-Type": "application/json" },
|
||||
});
|
||||
}
|
||||
|
||||
// Validate JWT token
|
||||
let claims;
|
||||
try {
|
||||
const secret = env.JWT_SECRET?.toString?.() || env.JWT_SECRET;
|
||||
if (!secret) {
|
||||
throw new Error("JWT_SECRET not configured");
|
||||
}
|
||||
claims = await verifyJwt(token, secret);
|
||||
} catch (err) {
|
||||
return new Response(JSON.stringify({ error: `Invalid token: ${err.message}` }), {
|
||||
status: 401,
|
||||
headers: { "Content-Type": "application/json" },
|
||||
});
|
||||
}
|
||||
|
||||
// Validate token claims match the request
|
||||
if (claims.cipher_id !== cipherId || claims.attachment_id !== attachmentId) {
|
||||
return new Response(JSON.stringify({ error: "Invalid download token" }), {
|
||||
status: 401,
|
||||
headers: { "Content-Type": "application/json" },
|
||||
});
|
||||
}
|
||||
|
||||
const userId = claims.sub;
|
||||
|
||||
// Verify cipher belongs to user and is not deleted
|
||||
const cipher = await db
|
||||
.prepare("SELECT * FROM ciphers WHERE id = ?1 AND user_id = ?2")
|
||||
.bind(cipherId, userId)
|
||||
.first();
|
||||
|
||||
if (!cipher) {
|
||||
return new Response(JSON.stringify({ error: "Cipher not found" }), {
|
||||
status: 404,
|
||||
headers: { "Content-Type": "application/json" },
|
||||
});
|
||||
}
|
||||
|
||||
if (cipher.organization_id) {
|
||||
return new Response(JSON.stringify({ error: "Organization attachments are not supported" }), {
|
||||
status: 400,
|
||||
headers: { "Content-Type": "application/json" },
|
||||
});
|
||||
}
|
||||
|
||||
if (cipher.deleted_at) {
|
||||
return new Response(
|
||||
JSON.stringify({ error: "Cannot modify attachments for deleted cipher" }),
|
||||
{ status: 400, headers: { "Content-Type": "application/json" } }
|
||||
);
|
||||
}
|
||||
|
||||
// Fetch pending attachment record
|
||||
const pending = await db
|
||||
.prepare("SELECT * FROM attachments_pending WHERE id = ?1")
|
||||
.bind(attachmentId)
|
||||
.first();
|
||||
|
||||
if (!pending) {
|
||||
return new Response(JSON.stringify({ error: "Attachment not found or already uploaded" }), {
|
||||
status: 404,
|
||||
headers: { "Content-Type": "application/json" },
|
||||
});
|
||||
}
|
||||
|
||||
if (pending.cipher_id !== cipherId) {
|
||||
return new Response(JSON.stringify({ error: "Attachment does not belong to cipher" }), {
|
||||
status: 400,
|
||||
headers: { "Content-Type": "application/json" },
|
||||
});
|
||||
}
|
||||
|
||||
// Get Content-Length from request headers
|
||||
const contentLengthHeader = request.headers.get("Content-Length");
|
||||
if (!contentLengthHeader) {
|
||||
return new Response(JSON.stringify({ error: "Missing Content-Length header" }), {
|
||||
status: 400,
|
||||
headers: { "Content-Type": "application/json" },
|
||||
});
|
||||
}
|
||||
|
||||
const contentLength = parseInt(contentLengthHeader, 10);
|
||||
if (isNaN(contentLength) || contentLength <= 0) {
|
||||
return new Response(JSON.stringify({ error: "Invalid Content-Length header" }), {
|
||||
status: 400,
|
||||
headers: { "Content-Type": "application/json" },
|
||||
});
|
||||
}
|
||||
|
||||
// Enforce limits before upload
|
||||
try {
|
||||
await enforceLimits(db, env, userId, contentLength, attachmentId);
|
||||
} catch (err) {
|
||||
return new Response(JSON.stringify({ error: err.message }), {
|
||||
status: 400,
|
||||
headers: { "Content-Type": "application/json" },
|
||||
});
|
||||
}
|
||||
|
||||
// Build R2 key
|
||||
const r2Key = `${cipherId}/${attachmentId}`;
|
||||
|
||||
// Prepare R2 put options
|
||||
const putOptions = {};
|
||||
const contentType = request.headers.get("Content-Type");
|
||||
if (contentType) {
|
||||
putOptions.httpMetadata = { contentType };
|
||||
}
|
||||
|
||||
// Upload to R2 directly using request.body (zero-copy streaming)
|
||||
let r2Object;
|
||||
try {
|
||||
r2Object = await bucket.put(r2Key, request.body, putOptions);
|
||||
} catch (err) {
|
||||
// Try to clean up on failure
|
||||
try {
|
||||
await bucket.delete(r2Key);
|
||||
} catch {
|
||||
// Ignore cleanup errors
|
||||
}
|
||||
return new Response(JSON.stringify({ error: `Upload failed: ${err.message}` }), {
|
||||
status: 500,
|
||||
headers: { "Content-Type": "application/json" },
|
||||
});
|
||||
}
|
||||
|
||||
const uploadedSize = r2Object.size;
|
||||
|
||||
// Verify uploaded size matches Content-Length
|
||||
if (uploadedSize !== contentLength) {
|
||||
try {
|
||||
await bucket.delete(r2Key);
|
||||
} catch {
|
||||
// Ignore cleanup errors
|
||||
}
|
||||
return new Response(JSON.stringify({ error: "Content-Length does not match uploaded size" }), {
|
||||
status: 400,
|
||||
headers: { "Content-Type": "application/json" },
|
||||
});
|
||||
}
|
||||
|
||||
// Finalize upload: move pending -> attachments and touch revision timestamps
|
||||
const now = nowString();
|
||||
await db.batch([
|
||||
db
|
||||
.prepare(
|
||||
"INSERT INTO attachments (id, cipher_id, file_name, file_size, akey, created_at, updated_at, organization_id) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)"
|
||||
)
|
||||
.bind(
|
||||
attachmentId,
|
||||
cipherId,
|
||||
pending.file_name,
|
||||
uploadedSize,
|
||||
pending.akey,
|
||||
pending.created_at || now,
|
||||
now,
|
||||
pending.organization_id || null
|
||||
),
|
||||
db.prepare("DELETE FROM attachments_pending WHERE id = ?1").bind(attachmentId),
|
||||
db.prepare("UPDATE ciphers SET updated_at = ?1 WHERE id = ?2").bind(now, cipherId),
|
||||
db.prepare("UPDATE users SET updated_at = ?1 WHERE id = ?2").bind(now, userId),
|
||||
]);
|
||||
|
||||
return new Response(null, { status: 201 });
|
||||
}
|
||||
|
||||
// Handle download with zero-copy streaming
|
||||
export async function handleDownload(request, env, cipherId, attachmentId, token) {
|
||||
// Get R2 bucket
|
||||
const bucket = env.ATTACHMENTS_BUCKET;
|
||||
if (!bucket) {
|
||||
return new Response(JSON.stringify({ error: "Attachments are not enabled" }), {
|
||||
status: 400,
|
||||
headers: { "Content-Type": "application/json" },
|
||||
});
|
||||
}
|
||||
|
||||
// Get D1 database
|
||||
const db = env.vault1;
|
||||
if (!db) {
|
||||
return new Response(JSON.stringify({ error: "Database not available" }), {
|
||||
status: 500,
|
||||
headers: { "Content-Type": "application/json" },
|
||||
});
|
||||
}
|
||||
|
||||
// Validate JWT token
|
||||
let claims;
|
||||
try {
|
||||
const secret = env.JWT_SECRET?.toString?.() || env.JWT_SECRET;
|
||||
if (!secret) {
|
||||
throw new Error("JWT_SECRET not configured");
|
||||
}
|
||||
claims = await verifyJwt(token, secret);
|
||||
} catch (err) {
|
||||
return new Response(JSON.stringify({ error: `Invalid token: ${err.message}` }), {
|
||||
status: 401,
|
||||
headers: { "Content-Type": "application/json" },
|
||||
});
|
||||
}
|
||||
|
||||
// Validate token claims match the request
|
||||
if (claims.cipher_id !== cipherId || claims.attachment_id !== attachmentId) {
|
||||
return new Response(JSON.stringify({ error: "Invalid download token" }), {
|
||||
status: 401,
|
||||
headers: { "Content-Type": "application/json" },
|
||||
});
|
||||
}
|
||||
|
||||
const userId = claims.sub;
|
||||
|
||||
// Verify cipher belongs to user
|
||||
const cipher = await db
|
||||
.prepare("SELECT * FROM ciphers WHERE id = ?1 AND user_id = ?2")
|
||||
.bind(cipherId, userId)
|
||||
.first();
|
||||
|
||||
if (!cipher) {
|
||||
return new Response(JSON.stringify({ error: "Cipher not found" }), {
|
||||
status: 404,
|
||||
headers: { "Content-Type": "application/json" },
|
||||
});
|
||||
}
|
||||
|
||||
// Fetch attachment record
|
||||
const attachment = await db
|
||||
.prepare("SELECT * FROM attachments WHERE id = ?1")
|
||||
.bind(attachmentId)
|
||||
.first();
|
||||
|
||||
if (!attachment) {
|
||||
return new Response(JSON.stringify({ error: "Attachment not found" }), {
|
||||
status: 404,
|
||||
headers: { "Content-Type": "application/json" },
|
||||
});
|
||||
}
|
||||
|
||||
if (attachment.cipher_id !== cipherId) {
|
||||
return new Response(JSON.stringify({ error: "Attachment does not belong to cipher" }), {
|
||||
status: 400,
|
||||
headers: { "Content-Type": "application/json" },
|
||||
});
|
||||
}
|
||||
|
||||
// Build R2 key
|
||||
const r2Key = `${cipherId}/${attachmentId}`;
|
||||
|
||||
// Get object from R2
|
||||
const r2Object = await bucket.get(r2Key);
|
||||
if (!r2Object) {
|
||||
return new Response(JSON.stringify({ error: "Attachment not found in storage" }), {
|
||||
status: 404,
|
||||
headers: { "Content-Type": "application/json" },
|
||||
});
|
||||
}
|
||||
|
||||
// Build response headers
|
||||
const headers = new Headers();
|
||||
const contentType = r2Object.httpMetadata?.contentType || "application/octet-stream";
|
||||
headers.set("Content-Type", contentType);
|
||||
headers.set("Content-Length", r2Object.size.toString());
|
||||
|
||||
// Return response with R2 object body directly - zero-copy streaming
|
||||
return new Response(r2Object.body, {
|
||||
status: 200,
|
||||
headers,
|
||||
});
|
||||
}
|
||||
|
||||
|
||||
460
src/entry.js
460
src/entry.js
|
|
@ -9,68 +9,16 @@
|
|||
* This avoids CPU time consumption that would occur if the body went through
|
||||
* the Rust/WASM layer with axum body conversion.
|
||||
*
|
||||
* Additionally, this wrapper can optionally offload CPU-heavy endpoints to a Rust Durable Object
|
||||
* (higher CPU budget) by enabling `HEAVY_DO_ENABLED=1` and binding `HEAVY_DO` in `wrangler.toml`.
|
||||
* This is used for operations like imports and password verification paths, keeping the main
|
||||
* Worker on a low-CPU fast path for typical requests.
|
||||
*
|
||||
* All other requests are passed through to the Rust WASM module.
|
||||
*/
|
||||
|
||||
import RustWorker from "../build/index.js";
|
||||
|
||||
// JWT validation using Web Crypto API (no external dependencies)
|
||||
async function verifyJWT(token, secret) {
|
||||
const encoder = new TextEncoder();
|
||||
const parts = token.split(".");
|
||||
if (parts.length !== 3) {
|
||||
throw new Error("Invalid token format");
|
||||
}
|
||||
|
||||
const [headerB64, payloadB64, signatureB64] = parts;
|
||||
|
||||
// Import the secret key for HMAC-SHA256
|
||||
const key = await crypto.subtle.importKey(
|
||||
"raw",
|
||||
encoder.encode(secret),
|
||||
{ name: "HMAC", hash: "SHA-256" },
|
||||
false,
|
||||
["verify"]
|
||||
);
|
||||
|
||||
// Decode the signature (base64url to Uint8Array)
|
||||
const signature = base64UrlDecode(signatureB64);
|
||||
|
||||
// Verify the signature
|
||||
const data = encoder.encode(`${headerB64}.${payloadB64}`);
|
||||
const valid = await crypto.subtle.verify("HMAC", key, signature, data);
|
||||
|
||||
if (!valid) {
|
||||
throw new Error("Invalid token signature");
|
||||
}
|
||||
|
||||
// Decode and parse the payload
|
||||
const payload = JSON.parse(
|
||||
new TextDecoder().decode(base64UrlDecode(payloadB64))
|
||||
);
|
||||
|
||||
// Check expiration
|
||||
if (payload.exp && payload.exp < Math.floor(Date.now() / 1000)) {
|
||||
throw new Error("Token expired");
|
||||
}
|
||||
|
||||
return payload;
|
||||
}
|
||||
|
||||
function base64UrlDecode(str) {
|
||||
// Convert base64url to base64
|
||||
let base64 = str.replace(/-/g, "+").replace(/_/g, "/");
|
||||
// Add padding if needed
|
||||
while (base64.length % 4) {
|
||||
base64 += "=";
|
||||
}
|
||||
const binary = atob(base64);
|
||||
const bytes = new Uint8Array(binary.length);
|
||||
for (let i = 0; i < binary.length; i++) {
|
||||
bytes[i] = binary.charCodeAt(i);
|
||||
}
|
||||
return bytes;
|
||||
}
|
||||
import { base64UrlDecode, handleAzureUpload, handleDownload } from "./attachments.js";
|
||||
|
||||
function isTruthy(value) {
|
||||
if (value == null) return false;
|
||||
|
|
@ -158,17 +106,6 @@ function parseDownloadPath(path) {
|
|||
return null;
|
||||
}
|
||||
|
||||
// Extract token from query string
|
||||
function extractTokenFromQuery(url) {
|
||||
const params = new URL(url).searchParams;
|
||||
return params.get("token");
|
||||
}
|
||||
|
||||
// Generate ISO timestamp string
|
||||
function nowString() {
|
||||
return new Date().toISOString();
|
||||
}
|
||||
|
||||
// Helper to get env var with fallback
|
||||
function getEnvVar(env, name, defaultValue = null) {
|
||||
try {
|
||||
|
|
@ -182,375 +119,6 @@ function getEnvVar(env, name, defaultValue = null) {
|
|||
}
|
||||
}
|
||||
|
||||
// Get attachment size limits from env
|
||||
function getAttachmentMaxBytes(env) {
|
||||
const value = getEnvVar(env, "ATTACHMENT_MAX_BYTES");
|
||||
if (!value) return null;
|
||||
const parsed = parseInt(value, 10);
|
||||
return isNaN(parsed) ? null : parsed;
|
||||
}
|
||||
|
||||
function getTotalLimitBytes(env) {
|
||||
const value = getEnvVar(env, "ATTACHMENT_TOTAL_LIMIT_KB");
|
||||
if (!value) return null;
|
||||
const kb = parseInt(value, 10);
|
||||
if (isNaN(kb)) return null;
|
||||
return kb * 1024;
|
||||
}
|
||||
|
||||
// Get user's current attachment usage
|
||||
async function getUserAttachmentUsage(db, userId, excludeAttachmentId) {
|
||||
const query = excludeAttachmentId
|
||||
? `SELECT COALESCE(SUM(file_size), 0) as total FROM (
|
||||
SELECT a.file_size as file_size
|
||||
FROM attachments a
|
||||
JOIN ciphers c ON c.id = a.cipher_id
|
||||
WHERE c.user_id = ?1 AND a.id != ?2
|
||||
UNION ALL
|
||||
SELECT p.file_size as file_size
|
||||
FROM attachments_pending p
|
||||
JOIN ciphers c2 ON c2.id = p.cipher_id
|
||||
WHERE c2.user_id = ?1 AND p.id != ?2
|
||||
) AS files`
|
||||
: `SELECT COALESCE(SUM(file_size), 0) as total FROM (
|
||||
SELECT a.file_size as file_size
|
||||
FROM attachments a
|
||||
JOIN ciphers c ON c.id = a.cipher_id
|
||||
WHERE c.user_id = ?1
|
||||
UNION ALL
|
||||
SELECT p.file_size as file_size
|
||||
FROM attachments_pending p
|
||||
JOIN ciphers c2 ON c2.id = p.cipher_id
|
||||
WHERE c2.user_id = ?1
|
||||
) AS files`;
|
||||
|
||||
const bindings = excludeAttachmentId
|
||||
? [userId, excludeAttachmentId]
|
||||
: [userId];
|
||||
|
||||
const result = await db
|
||||
.prepare(query)
|
||||
.bind(...bindings)
|
||||
.first();
|
||||
|
||||
return result?.total || 0;
|
||||
}
|
||||
|
||||
// Enforce attachment size limits
|
||||
async function enforceLimits(db, env, userId, newSize, excludeAttachmentId) {
|
||||
if (newSize < 0) {
|
||||
throw new Error("Attachment size cannot be negative");
|
||||
}
|
||||
|
||||
const maxBytes = getAttachmentMaxBytes(env);
|
||||
if (maxBytes !== null && newSize > maxBytes) {
|
||||
throw new Error("Attachment size exceeds limit");
|
||||
}
|
||||
|
||||
const limitBytes = getTotalLimitBytes(env);
|
||||
if (limitBytes !== null) {
|
||||
const used = await getUserAttachmentUsage(db, userId, excludeAttachmentId);
|
||||
const newTotal = used + newSize;
|
||||
if (newTotal > limitBytes) {
|
||||
throw new Error("Attachment storage limit reached");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Handle azure-upload with zero-copy streaming
|
||||
async function handleAzureUpload(request, env, cipherId, attachmentId, token) {
|
||||
// Get R2 bucket
|
||||
const bucket = env.ATTACHMENTS_BUCKET;
|
||||
if (!bucket) {
|
||||
return new Response(JSON.stringify({ error: "Attachments are not enabled" }), {
|
||||
status: 400,
|
||||
headers: { "Content-Type": "application/json" },
|
||||
});
|
||||
}
|
||||
|
||||
// Get D1 database
|
||||
const db = env.vault1;
|
||||
if (!db) {
|
||||
return new Response(JSON.stringify({ error: "Database not available" }), {
|
||||
status: 500,
|
||||
headers: { "Content-Type": "application/json" },
|
||||
});
|
||||
}
|
||||
|
||||
// Validate JWT token
|
||||
let claims;
|
||||
try {
|
||||
const secret = env.JWT_SECRET?.toString?.() || env.JWT_SECRET;
|
||||
if (!secret) {
|
||||
throw new Error("JWT_SECRET not configured");
|
||||
}
|
||||
claims = await verifyJWT(token, secret);
|
||||
} catch (err) {
|
||||
return new Response(JSON.stringify({ error: `Invalid token: ${err.message}` }), {
|
||||
status: 401,
|
||||
headers: { "Content-Type": "application/json" },
|
||||
});
|
||||
}
|
||||
|
||||
// Validate token claims match the request
|
||||
if (claims.cipher_id !== cipherId || claims.attachment_id !== attachmentId) {
|
||||
return new Response(JSON.stringify({ error: "Invalid download token" }), {
|
||||
status: 401,
|
||||
headers: { "Content-Type": "application/json" },
|
||||
});
|
||||
}
|
||||
|
||||
const userId = claims.sub;
|
||||
|
||||
// Verify cipher belongs to user and is not deleted
|
||||
const cipher = await db
|
||||
.prepare("SELECT * FROM ciphers WHERE id = ?1 AND user_id = ?2")
|
||||
.bind(cipherId, userId)
|
||||
.first();
|
||||
|
||||
if (!cipher) {
|
||||
return new Response(JSON.stringify({ error: "Cipher not found" }), {
|
||||
status: 404,
|
||||
headers: { "Content-Type": "application/json" },
|
||||
});
|
||||
}
|
||||
|
||||
if (cipher.organization_id) {
|
||||
return new Response(
|
||||
JSON.stringify({ error: "Organization attachments are not supported" }),
|
||||
{ status: 400, headers: { "Content-Type": "application/json" } }
|
||||
);
|
||||
}
|
||||
|
||||
if (cipher.deleted_at) {
|
||||
return new Response(
|
||||
JSON.stringify({ error: "Cannot modify attachments for deleted cipher" }),
|
||||
{ status: 400, headers: { "Content-Type": "application/json" } }
|
||||
);
|
||||
}
|
||||
|
||||
// Fetch pending attachment record
|
||||
const pending = await db
|
||||
.prepare("SELECT * FROM attachments_pending WHERE id = ?1")
|
||||
.bind(attachmentId)
|
||||
.first();
|
||||
|
||||
if (!pending) {
|
||||
return new Response(JSON.stringify({ error: "Attachment not found or already uploaded" }), {
|
||||
status: 404,
|
||||
headers: { "Content-Type": "application/json" },
|
||||
});
|
||||
}
|
||||
|
||||
if (pending.cipher_id !== cipherId) {
|
||||
return new Response(
|
||||
JSON.stringify({ error: "Attachment does not belong to cipher" }),
|
||||
{ status: 400, headers: { "Content-Type": "application/json" } }
|
||||
);
|
||||
}
|
||||
|
||||
// Get Content-Length from request headers
|
||||
const contentLengthHeader = request.headers.get("Content-Length");
|
||||
if (!contentLengthHeader) {
|
||||
return new Response(
|
||||
JSON.stringify({ error: "Missing Content-Length header" }),
|
||||
{ status: 400, headers: { "Content-Type": "application/json" } }
|
||||
);
|
||||
}
|
||||
|
||||
const contentLength = parseInt(contentLengthHeader, 10);
|
||||
if (isNaN(contentLength) || contentLength <= 0) {
|
||||
return new Response(
|
||||
JSON.stringify({ error: "Invalid Content-Length header" }),
|
||||
{ status: 400, headers: { "Content-Type": "application/json" } }
|
||||
);
|
||||
}
|
||||
|
||||
// Enforce limits before upload
|
||||
try {
|
||||
await enforceLimits(db, env, userId, contentLength, attachmentId);
|
||||
} catch (err) {
|
||||
return new Response(JSON.stringify({ error: err.message }), {
|
||||
status: 400,
|
||||
headers: { "Content-Type": "application/json" },
|
||||
});
|
||||
}
|
||||
|
||||
// Build R2 key
|
||||
const r2Key = `${cipherId}/${attachmentId}`;
|
||||
|
||||
// Prepare R2 put options
|
||||
const putOptions = {};
|
||||
const contentType = request.headers.get("Content-Type");
|
||||
if (contentType) {
|
||||
putOptions.httpMetadata = { contentType };
|
||||
}
|
||||
|
||||
// Upload to R2 directly using request.body
|
||||
// Workers R2 binding can accept request.body directly for zero-copy streaming
|
||||
// See: https://blog.cloudflare.com/zh-cn/r2-ga/
|
||||
let r2Object;
|
||||
try {
|
||||
r2Object = await bucket.put(r2Key, request.body, putOptions);
|
||||
} catch (err) {
|
||||
// Try to clean up on failure
|
||||
try {
|
||||
await bucket.delete(r2Key);
|
||||
} catch {
|
||||
// Ignore cleanup errors
|
||||
}
|
||||
return new Response(
|
||||
JSON.stringify({ error: `Upload failed: ${err.message}` }),
|
||||
{ status: 500, headers: { "Content-Type": "application/json" } }
|
||||
);
|
||||
}
|
||||
|
||||
const uploadedSize = r2Object.size;
|
||||
|
||||
// Verify uploaded size matches Content-Length
|
||||
if (uploadedSize !== contentLength) {
|
||||
try {
|
||||
await bucket.delete(r2Key);
|
||||
} catch {
|
||||
// Ignore cleanup errors
|
||||
}
|
||||
return new Response(
|
||||
JSON.stringify({ error: "Content-Length does not match uploaded size" }),
|
||||
{ status: 400, headers: { "Content-Type": "application/json" } }
|
||||
);
|
||||
}
|
||||
|
||||
// Finalize upload: move pending -> attachments and touch revision timestamps
|
||||
const now = nowString();
|
||||
await db.batch([
|
||||
db
|
||||
.prepare(
|
||||
"INSERT INTO attachments (id, cipher_id, file_name, file_size, akey, created_at, updated_at, organization_id) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)"
|
||||
)
|
||||
.bind(
|
||||
attachmentId,
|
||||
cipherId,
|
||||
pending.file_name,
|
||||
uploadedSize,
|
||||
pending.akey,
|
||||
pending.created_at || now,
|
||||
now,
|
||||
pending.organization_id || null
|
||||
),
|
||||
db
|
||||
.prepare("DELETE FROM attachments_pending WHERE id = ?1")
|
||||
.bind(attachmentId),
|
||||
db.prepare("UPDATE ciphers SET updated_at = ?1 WHERE id = ?2").bind(now, cipherId),
|
||||
db.prepare("UPDATE users SET updated_at = ?1 WHERE id = ?2").bind(now, userId),
|
||||
]);
|
||||
|
||||
return new Response(null, { status: 201 });
|
||||
}
|
||||
|
||||
// Handle download with zero-copy streaming
|
||||
async function handleDownload(request, env, cipherId, attachmentId, token) {
|
||||
// Get R2 bucket
|
||||
const bucket = env.ATTACHMENTS_BUCKET;
|
||||
if (!bucket) {
|
||||
return new Response(JSON.stringify({ error: "Attachments are not enabled" }), {
|
||||
status: 400,
|
||||
headers: { "Content-Type": "application/json" },
|
||||
});
|
||||
}
|
||||
|
||||
// Get D1 database
|
||||
const db = env.vault1;
|
||||
if (!db) {
|
||||
return new Response(JSON.stringify({ error: "Database not available" }), {
|
||||
status: 500,
|
||||
headers: { "Content-Type": "application/json" },
|
||||
});
|
||||
}
|
||||
|
||||
// Validate JWT token
|
||||
let claims;
|
||||
try {
|
||||
const secret = env.JWT_SECRET?.toString?.() || env.JWT_SECRET;
|
||||
if (!secret) {
|
||||
throw new Error("JWT_SECRET not configured");
|
||||
}
|
||||
claims = await verifyJWT(token, secret);
|
||||
} catch (err) {
|
||||
return new Response(JSON.stringify({ error: `Invalid token: ${err.message}` }), {
|
||||
status: 401,
|
||||
headers: { "Content-Type": "application/json" },
|
||||
});
|
||||
}
|
||||
|
||||
// Validate token claims match the request
|
||||
if (claims.cipher_id !== cipherId || claims.attachment_id !== attachmentId) {
|
||||
return new Response(JSON.stringify({ error: "Invalid download token" }), {
|
||||
status: 401,
|
||||
headers: { "Content-Type": "application/json" },
|
||||
});
|
||||
}
|
||||
|
||||
const userId = claims.sub;
|
||||
|
||||
// Verify cipher belongs to user
|
||||
const cipher = await db
|
||||
.prepare("SELECT * FROM ciphers WHERE id = ?1 AND user_id = ?2")
|
||||
.bind(cipherId, userId)
|
||||
.first();
|
||||
|
||||
if (!cipher) {
|
||||
return new Response(JSON.stringify({ error: "Cipher not found" }), {
|
||||
status: 404,
|
||||
headers: { "Content-Type": "application/json" },
|
||||
});
|
||||
}
|
||||
|
||||
// Fetch attachment record
|
||||
const attachment = await db
|
||||
.prepare("SELECT * FROM attachments WHERE id = ?1")
|
||||
.bind(attachmentId)
|
||||
.first();
|
||||
|
||||
if (!attachment) {
|
||||
return new Response(JSON.stringify({ error: "Attachment not found" }), {
|
||||
status: 404,
|
||||
headers: { "Content-Type": "application/json" },
|
||||
});
|
||||
}
|
||||
|
||||
if (attachment.cipher_id !== cipherId) {
|
||||
return new Response(
|
||||
JSON.stringify({ error: "Attachment does not belong to cipher" }),
|
||||
{ status: 400, headers: { "Content-Type": "application/json" } }
|
||||
);
|
||||
}
|
||||
|
||||
// Build R2 key
|
||||
const r2Key = `${cipherId}/${attachmentId}`;
|
||||
|
||||
// Get object from R2
|
||||
const r2Object = await bucket.get(r2Key);
|
||||
if (!r2Object) {
|
||||
return new Response(JSON.stringify({ error: "Attachment not found in storage" }), {
|
||||
status: 404,
|
||||
headers: { "Content-Type": "application/json" },
|
||||
});
|
||||
}
|
||||
|
||||
// Build response headers
|
||||
const headers = new Headers();
|
||||
const contentType = r2Object.httpMetadata?.contentType || "application/octet-stream";
|
||||
headers.set("Content-Type", contentType);
|
||||
headers.set("Content-Length", r2Object.size.toString());
|
||||
|
||||
// Return response with R2 object body directly - zero-copy streaming!
|
||||
// The Workers runtime streams the body directly to the client without consuming CPU time
|
||||
return new Response(r2Object.body, {
|
||||
status: 200,
|
||||
headers,
|
||||
});
|
||||
}
|
||||
|
||||
// Main fetch handler
|
||||
export default {
|
||||
async fetch(request, env, ctx) {
|
||||
|
|
@ -562,6 +130,7 @@ export default {
|
|||
// Import: route to Rust Durable Object (HeavyDo) to reuse the existing Rust import logic.
|
||||
if (request.method === "POST" && url.pathname === "/api/ciphers/import") {
|
||||
if (!env.HEAVY_DO) {
|
||||
console.error("HEAVY_DO binding not configured");
|
||||
return new Response(JSON.stringify({ error: "HEAVY_DO binding not configured" }), {
|
||||
status: 500,
|
||||
headers: { "Content-Type": "application/json" },
|
||||
|
|
@ -611,12 +180,17 @@ export default {
|
|||
}
|
||||
}
|
||||
|
||||
// Intercept PUT requests to azure-upload endpoint
|
||||
// Attachment upload/download fast-path (R2 zero-copy streaming + JWT validation)
|
||||
if (request.method === "PUT") {
|
||||
const parsed = parseAzureUploadPath(url.pathname);
|
||||
if (parsed) {
|
||||
const token = extractTokenFromQuery(request.url);
|
||||
if (token) {
|
||||
const token = url.searchParams.get("token");
|
||||
if (!token) {
|
||||
return new Response(
|
||||
JSON.stringify({ error: "Missing upload token" }),
|
||||
{ status: 401, headers: { "Content-Type": "application/json" } }
|
||||
);
|
||||
}
|
||||
return handleAzureUpload(
|
||||
request,
|
||||
env,
|
||||
|
|
@ -626,13 +200,11 @@ export default {
|
|||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Intercept GET requests to download endpoint for zero-copy streaming
|
||||
if (request.method === "GET") {
|
||||
const parsed = parseDownloadPath(url.pathname);
|
||||
if (parsed) {
|
||||
const token = extractTokenFromQuery(request.url);
|
||||
const token = url.searchParams.get("token");
|
||||
if (!token) {
|
||||
return new Response(
|
||||
JSON.stringify({ error: "Missing download token" }),
|
||||
|
|
|
|||
Loading…
Reference in a new issue