7326e273e1
Protect upload cleanup and atomic completion paths, add durable media worker leases and brand asset validation, and introduce verified migration/bootstrap/deployment gates.
378 lines
11 KiB
TypeScript
378 lines
11 KiB
TypeScript
import { S3Client, PutObjectCommand, DeleteObjectCommand, GetObjectCommand, HeadObjectCommand } from '@aws-sdk/client-s3'
|
||
import { createReadStream } from 'fs'
|
||
import { lstat } from 'fs/promises'
|
||
import { Readable } from 'stream'
|
||
|
||
/**
|
||
* 从 S3 公网 URL 或 MinIO path-style URL 中提取对象 Key。
|
||
*/
|
||
export function getS3KeyFromUrl(fileUrl: string): string {
|
||
const url = new URL(fileUrl)
|
||
let key = decodeURIComponent(url.pathname.slice(1))
|
||
const bucket = process.env.S3_BUCKET || ''
|
||
|
||
if (bucket && key.startsWith(`${bucket}/`)) {
|
||
key = key.slice(bucket.length + 1)
|
||
}
|
||
|
||
return key
|
||
}
|
||
|
||
function decodeObjectKeyForValidation(value: string): string {
|
||
let decoded = value
|
||
for (let index = 0; index < 5; index += 1) {
|
||
try {
|
||
const next = decodeURIComponent(decoded)
|
||
if (next === decoded) break
|
||
decoded = next
|
||
} catch {
|
||
break
|
||
}
|
||
}
|
||
return decoded
|
||
}
|
||
|
||
function normalizeObjectPrefix(value: string | undefined): string {
|
||
return String(value || '')
|
||
.trim()
|
||
.replace(/\\/g, '/')
|
||
.replace(/^\/+|\/+$/g, '')
|
||
}
|
||
|
||
/**
|
||
* 删除属于高风险写操作，只接受当前配置的公网域名或 S3 Endpoint。
|
||
* 同时限制到当前业务前缀，拒绝把任意 HTTP URL 当作本桶对象 Key。
|
||
*/
|
||
export function getTrustedS3MutationKey(fileUrl: string): string {
|
||
const target = new URL(fileUrl)
|
||
if (
|
||
!['http:', 'https:'].includes(target.protocol) ||
|
||
target.username ||
|
||
target.password
|
||
) {
|
||
throw new Error('拒绝不受信任的对象存储地址')
|
||
}
|
||
|
||
const configuredUrls = [
|
||
process.env.S3_PUBLIC_DOMAIN,
|
||
cleanS3Endpoint(process.env.S3_ENDPOINT),
|
||
]
|
||
.map((value) => String(value || '').trim())
|
||
.filter(Boolean)
|
||
.map((value) => new URL(value))
|
||
|
||
const trustedBase = configuredUrls.find((configured) => {
|
||
if (configured.origin !== target.origin) {
|
||
return false
|
||
}
|
||
|
||
const basePath = configured.pathname.replace(/\/+$/g, '')
|
||
return (
|
||
!basePath ||
|
||
basePath === '/' ||
|
||
target.pathname === basePath ||
|
||
target.pathname.startsWith(`${basePath}/`)
|
||
)
|
||
})
|
||
if (!trustedBase) {
|
||
throw new Error('拒绝不受信任的对象存储地址')
|
||
}
|
||
|
||
const key = decodeObjectKeyForValidation(getS3KeyFromUrl(fileUrl))
|
||
.replace(/^\/+/g, '')
|
||
const keySegments = key.split('/')
|
||
if (
|
||
!key ||
|
||
key.includes('\\') ||
|
||
/[\x00-\x1f\x7f]/.test(key) ||
|
||
keySegments.some((segment) => !segment || segment === '.' || segment === '..')
|
||
) {
|
||
throw new Error('拒绝不安全的对象存储 Key')
|
||
}
|
||
|
||
const requiredPrefix = normalizeObjectPrefix(process.env.S3_FILE_PATH)
|
||
if (
|
||
requiredPrefix &&
|
||
key !== requiredPrefix &&
|
||
!key.startsWith(`${requiredPrefix}/`)
|
||
) {
|
||
throw new Error('拒绝删除当前业务前缀之外的对象')
|
||
}
|
||
|
||
return key
|
||
}
|
||
|
||
/**
|
||
* 判断 URL 是否为 HTTP/HTTPS URL
|
||
*/
|
||
export function isHttpUrl(url: string): boolean {
|
||
if (!url) return false
|
||
return /^https?:\/\//.test(url)
|
||
}
|
||
|
||
/**
|
||
* 判断 URL 是否为 S3 URL
|
||
*/
|
||
export function isS3Url(url: string): boolean {
|
||
if (!isHttpUrl(url)) return false
|
||
try {
|
||
getTrustedS3MutationKey(url)
|
||
return true
|
||
} catch {
|
||
return false
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 判断 URL 是否为 S3 公网 URL
|
||
*/
|
||
export function isS3PublicUrl(url: string): boolean {
|
||
if (!url || !isHttpUrl(url)) return false
|
||
const publicDomain = process.env.S3_PUBLIC_DOMAIN || ''
|
||
if (!publicDomain) return false
|
||
|
||
try {
|
||
const urlObj = new URL(url)
|
||
const domainObj = new URL(publicDomain)
|
||
return urlObj.hostname === domainObj.hostname
|
||
} catch {
|
||
return false
|
||
}
|
||
}
|
||
|
||
// 清理S3_ENDPOINT，去除可能的注释和空白字符
|
||
function cleanS3Endpoint(endpoint: string | undefined): string {
|
||
if (!endpoint) return ''
|
||
// 去除注释（# 后面的内容）和前后空白
|
||
const cleaned = endpoint.split('#')[0].trim()
|
||
return cleaned
|
||
}
|
||
|
||
/**
|
||
* R6修复：优化 S3 连接配置
|
||
*
|
||
* 原配置问题：
|
||
* - maxAttempts: 10 太激进，网络问题时可能长时间阻塞
|
||
* - 缺少请求级别的超时配置
|
||
*
|
||
* 优化后：
|
||
* - maxAttempts: 5（平衡重试次数和响应时间）
|
||
* - 保留 adaptive 重试模式（自动调整重试间隔）
|
||
* - 添加更详细的日志记录
|
||
*/
|
||
const s3Client = new S3Client({
|
||
endpoint: cleanS3Endpoint(process.env.S3_ENDPOINT),
|
||
region: process.env.S3_REGION || 'us-east-1',
|
||
credentials: {
|
||
accessKeyId: process.env.S3_ACCESS_KEY_ID!,
|
||
secretAccessKey: process.env.S3_SECRET_ACCESS_KEY!,
|
||
},
|
||
forcePathStyle: true, // MinIO requires path-style
|
||
requestHandler: {
|
||
// 超时配置
|
||
requestTimeout: 300000, // 5分钟请求超时（大文件上传需要）
|
||
connectionTimeout: 30000, // 30秒连接超时
|
||
},
|
||
// R6修复：将重试次数从10次降低到5次
|
||
// 理由：10次重试 + adaptive 模式可能导致单次请求阻塞数分钟
|
||
// 5次重试足以应对临时网络抖动，同时避免长时间阻塞
|
||
maxAttempts: 5,
|
||
retryMode: 'adaptive', // 自动调整重试间隔（指数退避 + 抖动）
|
||
})
|
||
|
||
export async function uploadToS3(
|
||
file: Buffer,
|
||
fileName: string,
|
||
contentType: string
|
||
): Promise<string> {
|
||
// 确保 S3_FILE_PATH 以 / 结尾，fileName 不以 / 开头
|
||
const s3FilePath = (process.env.S3_FILE_PATH || '').trim()
|
||
const normalizedPath = s3FilePath && !s3FilePath.endsWith('/') ? `${s3FilePath}/` : s3FilePath
|
||
const normalizedFileName = fileName.startsWith('/') ? fileName.slice(1) : fileName
|
||
const key = `${normalizedPath}${normalizedFileName}`
|
||
|
||
console.log(`[S3] Uploading file: key=${key}, size=${file.length}, contentType=${contentType}`)
|
||
|
||
const command = new PutObjectCommand({
|
||
Bucket: process.env.S3_BUCKET!,
|
||
Key: key,
|
||
Body: file,
|
||
ContentType: contentType,
|
||
ACL: 'public-read',
|
||
})
|
||
|
||
try {
|
||
await s3Client.send(command)
|
||
const url = `${process.env.S3_PUBLIC_DOMAIN}/${key}`
|
||
console.log(`[S3] Uploaded successfully: url=${url}`)
|
||
return url
|
||
} catch (error) {
|
||
// S3上传失败会自动重试（maxAttempts: 10），这里只记录日志
|
||
console.error('[S3] Upload error (after all retries):', error)
|
||
// 重新抛出错误，让上层处理
|
||
throw new Error('S3上传失败，请稍后重试')
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 上传完成后、写数据库前确认对象确实存在且元数据与本次文件一致。
|
||
* 验证失败时调用方不得覆盖原数据库地址；可能产生的孤立对象交由
|
||
* 后续“先核对数据库引用再清理”的任务处理，避免误删有效对象。
|
||
*/
|
||
export async function verifyUploadedS3Object(
|
||
fileUrl: string,
|
||
expectedSize: number,
|
||
expectedContentType: string,
|
||
): Promise<void> {
|
||
const key = getTrustedS3MutationKey(fileUrl)
|
||
const response = await s3Client.send(new HeadObjectCommand({
|
||
Bucket: process.env.S3_BUCKET!,
|
||
Key: key,
|
||
}))
|
||
|
||
if (response.ContentLength !== expectedSize) {
|
||
throw new Error('上传对象大小校验失败')
|
||
}
|
||
|
||
const actualContentType = String(response.ContentType || '')
|
||
.split(';', 1)[0]
|
||
.trim()
|
||
.toLowerCase()
|
||
if (actualContentType !== expectedContentType.toLowerCase()) {
|
||
throw new Error('上传对象类型校验失败')
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 将磁盘文件以流的方式写入 S3，避免把视频或图片再次整体读入 Node.js 内存。
|
||
*/
|
||
export async function uploadFileToS3(
|
||
filePath: string,
|
||
fileName: string,
|
||
contentType: string,
|
||
expectedSize?: number,
|
||
): Promise<string> {
|
||
const fileStats = await lstat(filePath)
|
||
if (fileStats.isSymbolicLink() || !fileStats.isFile() || fileStats.size <= 0) {
|
||
throw new Error('待上传文件无效')
|
||
}
|
||
if (expectedSize !== undefined && (
|
||
!Number.isSafeInteger(expectedSize)
|
||
|| expectedSize <= 0
|
||
|| fileStats.size !== expectedSize
|
||
)) {
|
||
throw new Error('待上传文件实际大小与上传记录不一致')
|
||
}
|
||
|
||
const s3FilePath = (process.env.S3_FILE_PATH || '').trim()
|
||
const normalizedPath = s3FilePath && !s3FilePath.endsWith('/') ? `${s3FilePath}/` : s3FilePath
|
||
const normalizedFileName = fileName.startsWith('/') ? fileName.slice(1) : fileName
|
||
const key = `${normalizedPath}${normalizedFileName}`
|
||
const body = createReadStream(filePath)
|
||
const command = new PutObjectCommand({
|
||
Bucket: process.env.S3_BUCKET!,
|
||
Key: key,
|
||
Body: body,
|
||
ContentLength: fileStats.size,
|
||
ContentType: contentType,
|
||
ACL: 'public-read',
|
||
})
|
||
|
||
try {
|
||
await s3Client.send(command)
|
||
return `${process.env.S3_PUBLIC_DOMAIN}/${key}`
|
||
} catch (error) {
|
||
body.destroy()
|
||
console.error('[S3] File stream upload error (after all retries):', error)
|
||
throw new Error('S3上传失败，请稍后重试')
|
||
}
|
||
}
|
||
|
||
export async function downloadFromS3(fileUrl: string): Promise<Buffer> {
|
||
try {
|
||
const key = getS3KeyFromUrl(fileUrl)
|
||
const bucket = process.env.S3_BUCKET!
|
||
|
||
console.log(`[S3] Downloading file: bucket=${bucket}, key=${key}`)
|
||
|
||
const command = new GetObjectCommand({
|
||
Bucket: bucket,
|
||
Key: key,
|
||
})
|
||
|
||
const response = await s3Client.send(command)
|
||
|
||
// Convert stream to buffer
|
||
if (response.Body instanceof Readable) {
|
||
const chunks: Buffer[] = []
|
||
for await (const chunk of response.Body as Readable) {
|
||
chunks.push(Buffer.from(chunk))
|
||
}
|
||
const buffer = Buffer.concat(chunks)
|
||
console.log(`[S3] Downloaded successfully: size=${buffer.length}`)
|
||
return buffer
|
||
}
|
||
|
||
throw new Error('Invalid response body')
|
||
} catch (error) {
|
||
console.error('[S3] Download error:', error)
|
||
throw new Error('从S3下载文件失败')
|
||
}
|
||
}
|
||
|
||
export async function deleteFromS3(fileUrl: string): Promise<void> {
|
||
try {
|
||
const key = getTrustedS3MutationKey(fileUrl)
|
||
|
||
const command = new DeleteObjectCommand({
|
||
Bucket: process.env.S3_BUCKET!,
|
||
Key: key,
|
||
})
|
||
|
||
await s3Client.send(command)
|
||
} catch (error) {
|
||
console.error('Error deleting from S3:', error)
|
||
throw error
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 从 S3 获取对象流
|
||
*/
|
||
export async function getS3ObjectStream(key: string): Promise<{ body: Readable; contentLength?: number }> {
|
||
const command = new GetObjectCommand({
|
||
Bucket: process.env.S3_BUCKET!,
|
||
Key: key,
|
||
})
|
||
|
||
const response = await s3Client.send(command)
|
||
return {
|
||
body: response.Body as Readable,
|
||
contentLength: response.ContentLength,
|
||
}
|
||
}
|
||
|
||
export { s3Client }
|
||
|
||
/**
|
||
* 测试 S3 连接
|
||
*/
|
||
export async function testS3Connection(): Promise<boolean> {
|
||
try {
|
||
// 尝试列出bucket（不需要实际列出，只是测试连接）
|
||
const command = new GetObjectCommand({
|
||
Bucket: process.env.S3_BUCKET!,
|
||
Key: 'test-connection', // 不存在的key，只是测试连接
|
||
})
|
||
|
||
await s3Client.send(command).catch(() => {
|
||
// 预期会失败（key不存在），但能连接就说明S3服务正常
|
||
})
|
||
|
||
return true
|
||
} catch (error) {
|
||
console.error('[S3] Connection test failed:', error)
|
||
return false
|
||
}
|
||
}
|