CDK(Cloud Development Kit)でファイルアップローダーを作ってみた。アップロードされた画像・動画・音声・PDF・テキスト(html)は、別タブで表示可能(ブラウザでそのまま表示できる)

ブラウザ ⇔ index.html＠S3 ⇔ API Gateway ⇔ lambda(POST,GET) ⇔ アップロードされたファイル＠S3

アップロードされた画像・動画・音声・PDF・テキスト(html)は、別タブで表示可能(ブラウザでそのまま表示できる)
エクセルファイルなど、ブラウザで表示できない場合はダウンロードされます。

mkdir mini-cdk-uploader && cd mini-cdk-uploader
cdk init app --language typescript
npm i aws-cdk-lib constructs

mkdir -p src/app
touch src/app/handler.py

mkdir -p frontend
touch frontend/index.html

mkdir mini-cdk-uploader && cd mini-cdk-uploader

cdk init app --language typescript

npm i aws-cdk-lib constructs

mkdir -p src/app

touch src/app/handler.py

mkdir -p frontend

touch frontend/index.html

インフラ生成、
mini-cdk-uploader-stack.ts

// AWS CDKライブラリから必要なクラスをインポート
import { Stack, StackProps, Duration, CfnOutput } from "aws-cdk-lib";
// Construct（CDKの基本構成要素）をインポート
import { Construct } from "constructs";
// S3関連のCDKモジュールをインポート
import * as s3 from "aws-cdk-lib/aws-s3";
// S3へ静的ファイルをデプロイするためのモジュールをインポート
import * as s3deploy from "aws-cdk-lib/aws-s3-deployment";
// Lambda関数を作成するためのモジュールをインポート
import * as lambda from "aws-cdk-lib/aws-lambda";
// API Gatewayを作成するためのモジュールをインポート
import * as apigw from "aws-cdk-lib/aws-apigateway";
// IAM（権限設定）を扱うモジュールをインポート
import * as iam from "aws-cdk-lib/aws-iam";
// パス操作用のNode.js標準モジュール
import * as path from "path";

// mini-cdk-uploaderのスタック定義クラス
export class MiniCdkUploaderStack extends Stack {
  constructor(scope: Construct, id: string, props?: StackProps) {
    super(scope, id, props);

    // =====================================
    // 1. アップロード先のS3バケット（非公開）
    // =====================================
    const uploadBucket = new s3.Bucket(this, "UploadBucket", {
      blockPublicAccess: s3.BlockPublicAccess.BLOCK_ALL, // 外部公開をブロック
      encryption: s3.BucketEncryption.S3_MANAGED, // S3管理の暗号化を有効化
      enforceSSL: true, // HTTPS接続を強制
      // ブラウザから直接PUTできるようにCORSを設定
      cors: [
        {
          allowedMethods: [
            s3.HttpMethods.GET, // GET許可（ダウンロード）
            s3.HttpMethods.PUT, // PUT許可（アップロード）
            s3.HttpMethods.HEAD, // HEAD許可（プリフライト対策）
          ],
          allowedOrigins: ["*"], // どのオリジンからも許可（デモ用）
          allowedHeaders: ["*"], // 任意のヘッダーを許可
          exposedHeaders: ["ETag"], // レスポンスで公開するヘッダー
          maxAge: 3600, // プリフライトキャッシュ時間（秒）
        },
      ],
    });

    // =====================================
    // 2. フロントエンド用のS3バケット（静的サイトホスティング）
    // =====================================
    const siteBucket = new s3.Bucket(this, "SiteBucket", {
      websiteIndexDocument: "index.html", // サイトのルートドキュメント
      publicReadAccess: true, // 誰でも閲覧可（デモ用途）
      // パブリックアクセスを許可するための設定（OAIなし構成）
      blockPublicAccess: new s3.BlockPublicAccess({
        blockPublicAcls: false,
        blockPublicPolicy: false,
        ignorePublicAcls: false,
        restrictPublicBuckets: false,
      }),
    });

    // =====================================
    // 3. Lambda関数（API用バックエンド）
    // =====================================
    const handler = new lambda.Function(this, "ApiHandler", {
      runtime: lambda.Runtime.PYTHON_3_12, // 実行環境（Python 3.12）
      handler: "handler.handler", // エントリーポイント
      code: lambda.Code.fromAsset(path.join(__dirname, "../src/app")), // ソースコードのパス
      timeout: Duration.seconds(10), // タイムアウト設定
      environment: {
        UPLOAD_BUCKET: uploadBucket.bucketName, // 環境変数（バケット名）
        URL_EXPIRES: "3600", // 署名付きURLの有効期限（秒）
      },
    });

    // LambdaにS3アクセス権限を付与（PUT/GET/LISTなど）
    uploadBucket.grantPut(handler);
    uploadBucket.grantRead(handler);
    uploadBucket.grantReadWrite(handler); // list含む広めの権限

    // =====================================
    // 4. API Gateway（LambdaをHTTPで呼び出す）
    // =====================================
    const api = new apigw.RestApi(this, "UploaderApi", {
      restApiName: "mini-cdk-uploader-api", // APIの表示名
      defaultCorsPreflightOptions: {
        allowOrigins: apigw.Cors.ALL_ORIGINS, // すべてのオリジンからのCORS許可（デモ用）
        allowHeaders: apigw.Cors.DEFAULT_HEADERS, // 標準ヘッダー許可
        allowMethods: ["GET", "POST", "OPTIONS"], // 利用可能なHTTPメソッド
      },
      deployOptions: {
        stageName: "prod", // ステージ名
      },
    });

    // /presign エンドポイント：ファイルアップロード用プリサインURLを返す
    const presign = api.root.addResource("presign");
    presign.addMethod("POST", new apigw.LambdaIntegration(handler));

    // /list エンドポイント：アップロード済みファイル一覧を返す
    const list = api.root.addResource("list");
    list.addMethod("GET", new apigw.LambdaIntegration(handler));

    // =====================================
    // 5. フロントエンド(index.htmlなど)をS3へデプロイ
    // =====================================
    new s3deploy.BucketDeployment(this, "DeployFrontend", {
      destinationBucket: siteBucket, // デプロイ先
      sources: [s3deploy.Source.asset(path.join(__dirname, "../frontend"))], // ローカルのfrontendディレクトリ
    });

    // =====================================
    // 6. スタック出力（デプロイ後に確認しやすく）
    // =====================================
    new CfnOutput(this, "WebsiteURL", { value: siteBucket.bucketWebsiteUrl }); // フロントエンドURL
    new CfnOutput(this, "ApiBaseURL", { value: api.url }); // APIのエンドポイントURL
    new CfnOutput(this, "UploadBucketName", { value: uploadBucket.bucketName }); // バケット名
  }
}

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

// AWS CDKライブラリから必要なクラスをインポート

import { Stack, StackProps, Duration, CfnOutput } from "aws-cdk-lib";

// Construct（CDKの基本構成要素）をインポート

import { Construct } from "constructs";

// S3関連のCDKモジュールをインポート

import * as s3 from "aws-cdk-lib/aws-s3";

// S3へ静的ファイルをデプロイするためのモジュールをインポート

import * as s3deploy from "aws-cdk-lib/aws-s3-deployment";

// Lambda関数を作成するためのモジュールをインポート

import * as lambda from "aws-cdk-lib/aws-lambda";

// API Gatewayを作成するためのモジュールをインポート

import * as apigw from "aws-cdk-lib/aws-apigateway";

// IAM（権限設定）を扱うモジュールをインポート

import * as iam from "aws-cdk-lib/aws-iam";

// パス操作用のNode.js標準モジュール

import * as path from "path";

// mini-cdk-uploaderのスタック定義クラス

export class MiniCdkUploaderStack extends Stack {

constructor(scope: Construct, id: string, props?: StackProps) {

super(scope, id, props);

// =====================================

// 1. アップロード先のS3バケット（非公開）

// =====================================

const uploadBucket = new s3.Bucket(this, "UploadBucket", {

blockPublicAccess: s3.BlockPublicAccess.BLOCK_ALL, // 外部公開をブロック

encryption: s3.BucketEncryption.S3_MANAGED, // S3管理の暗号化を有効化

enforceSSL: true, // HTTPS接続を強制

// ブラウザから直接PUTできるようにCORSを設定

cors: [

{

allowedMethods: [

s3.HttpMethods.GET, // GET許可（ダウンロード）

s3.HttpMethods.PUT, // PUT許可（アップロード）

s3.HttpMethods.HEAD, // HEAD許可（プリフライト対策）

allowedOrigins: ["*"], // どのオリジンからも許可（デモ用）

allowedHeaders: ["*"], // 任意のヘッダーを許可

exposedHeaders: ["ETag"], // レスポンスで公開するヘッダー

maxAge: 3600, // プリフライトキャッシュ時間（秒）

});

// =====================================

// 2. フロントエンド用のS3バケット（静的サイトホスティング）

// =====================================

const siteBucket = new s3.Bucket(this, "SiteBucket", {

websiteIndexDocument: "index.html", // サイトのルートドキュメント

publicReadAccess: true, // 誰でも閲覧可（デモ用途）

// パブリックアクセスを許可するための設定（OAIなし構成）

blockPublicAccess: new s3.BlockPublicAccess({

blockPublicAcls: false,

blockPublicPolicy: false,

ignorePublicAcls: false,

restrictPublicBuckets: false,

}),

});

// =====================================

// 3. Lambda関数（API用バックエンド）

// =====================================

const handler = new lambda.Function(this, "ApiHandler", {

runtime: lambda.Runtime.PYTHON_3_12, // 実行環境（Python 3.12）

handler: "handler.handler", // エントリーポイント

code: lambda.Code.fromAsset(path.join(__dirname, "../src/app")), // ソースコードのパス

timeout: Duration.seconds(10), // タイムアウト設定

environment: {

UPLOAD_BUCKET: uploadBucket.bucketName, // 環境変数（バケット名）

URL_EXPIRES: "3600", // 署名付きURLの有効期限（秒）

});

// LambdaにS3アクセス権限を付与（PUT/GET/LISTなど）

uploadBucket.grantPut(handler);

uploadBucket.grantRead(handler);

uploadBucket.grantReadWrite(handler); // list含む広めの権限

// =====================================

// 4. API Gateway（LambdaをHTTPで呼び出す）

// =====================================

const api = new apigw.RestApi(this, "UploaderApi", {

restApiName: "mini-cdk-uploader-api", // APIの表示名

defaultCorsPreflightOptions: {

allowOrigins: apigw.Cors.ALL_ORIGINS, // すべてのオリジンからのCORS許可（デモ用）

allowHeaders: apigw.Cors.DEFAULT_HEADERS, // 標準ヘッダー許可

allowMethods: ["GET", "POST", "OPTIONS"], // 利用可能なHTTPメソッド

deployOptions: {

stageName: "prod", // ステージ名

});

// /presign エンドポイント：ファイルアップロード用プリサインURLを返す

const presign = api.root.addResource("presign");

presign.addMethod("POST", new apigw.LambdaIntegration(handler));

// /list エンドポイント：アップロード済みファイル一覧を返す

const list = api.root.addResource("list");

list.addMethod("GET", new apigw.LambdaIntegration(handler));

// =====================================

// 5. フロントエンド(index.htmlなど)をS3へデプロイ

// =====================================

new s3deploy.BucketDeployment(this, "DeployFrontend", {

destinationBucket: siteBucket, // デプロイ先

sources: [s3deploy.Source.asset(path.join(__dirname, "../frontend"))], // ローカルのfrontendディレクトリ

});

// =====================================

// 6. スタック出力（デプロイ後に確認しやすく）

// =====================================

new CfnOutput(this, "WebsiteURL", { value: siteBucket.bucketWebsiteUrl }); // フロントエンドURL

new CfnOutput(this, "ApiBaseURL", { value: api.url }); // APIのエンドポイントURL

new CfnOutput(this, "UploadBucketName", { value: uploadBucket.bucketName }); // バケット名

}

S3へのアップロード & アップロード済みファイル一覧取得
src/app/handler.py

# -*- coding: utf-8 -*-
"""
API Gateway + Lambda (Python) 用のアップローダーAPIハンドラ（強化版）

変更要点:
- 「表示」リンクで拡張子が嘘でも正しく表示できるように、
  Content-Type 決定ロジックを強化。
  優先順位:
    1) x-amz-meta-ctype（アップロード時に記録）
    2) オブジェクト先頭バイトのマジックナンバー検出（PDF/PNG/JPEG/GIF/MP4/WebM/MP3/WAV/SVG）
    3) S3のContentType
    4) 拡張子推定（mimetypes）
- 安全に inline できるタイプのみ inline。その他は attachment。
- 日本語ファイル名のダウンロード対応（filename*）
"""

import json
import os
import re
import urllib.parse
import mimetypes
from io import BytesIO

import boto3
from botocore.exceptions import ClientError
from botocore.config import Config

# ---------- 設定 ----------
s3 = boto3.client("s3", config=Config(s3={"addressing_style": "virtual"}))
BUCKET = os.environ.get("UPLOAD_BUCKET", "")
EXPIRES = int(os.environ.get("URL_EXPIRES", "3600"))

CORS_HEADERS = {
    "Content-Type": "application/json",
    "Access-Control-Allow-Origin": "*",
    "Access-Control-Allow-Headers": "Content-Type,Authorization",
    "Access-Control-Allow-Methods": "GET,POST,OPTIONS",
}

SAFE_INLINE_PREFIXES = ("image/", "video/", "audio/", "text/")
SAFE_INLINE_EXACT = {"application/pdf"}

# ---------- ユーティリティ ----------
def _response(status: int, body) -> dict:
    return {
        "statusCode": status,
        "headers": CORS_HEADERS,
        "body": json.dumps(body, ensure_ascii=False),
    }

def _ascii_fallback(name: str) -> str:
    if "." in name:
        stem, ext = name.rsplit(".", 1)
        safe_stem = re.sub(r"[^A-Za-z0-9._-]", "_", stem) or "download"
        return f"{safe_stem}.{ext}"
    else:
        return re.sub(r"[^A-Za-z0-9._-]", "_", name) or "download"

def _disp_attachment(filename: str) -> str:
    fallback = _ascii_fallback(filename)
    encoded = urllib.parse.quote(filename, safe="", encoding="utf-8")
    return f'attachment; filename="{fallback}"; filename*=UTF-8\'\'{encoded}'

def _guess_content_type(key: str) -> str:
    guess, _ = mimetypes.guess_type(key)
    return guess or "application/octet-stream"

def _is_safe_inline(ct: str) -> bool:
    return (ct in SAFE_INLINE_EXACT) or any(ct.startswith(p) for p in SAFE_INLINE_PREFIXES)

def _read_magic(bucket: str, key: str, length: int = 2048) -> bytes:
    """オブジェクト先頭を部分取得して返す。失敗時は空バイト列。"""
    try:
        resp = s3.get_object(Bucket=bucket, Key=key, Range=f"bytes=0-{max(0,length-1)}")
        return resp["Body"].read(length)
    except ClientError:
        return b""

def _sniff_by_magic(sample: bytes) -> str | None:
    """先頭バイトから代表的なContent-Typeを推定。該当なしならNone。"""
    if not sample:
        return None

    # PDF: %PDF-
    if sample.startswith(b"%PDF-"):
        return "application/pdf"
    # PNG
    if sample.startswith(b"\x89PNG\r\n\x1a\n"):
        return "image/png"
    # JPEG
    if sample.startswith(b"\xFF\xD8\xFF"):
        return "image/jpeg"
    # GIF
    if sample.startswith(b"GIF87a") or sample.startswith(b"GIF89a"):
        return "image/gif"
    # WebP (RIFF....WEBP)
    if sample.startswith(b"RIFF") and b"WEBP" in sample[8:16]:
        return "image/webp"
    # SVG (テキストXML、先頭に <svg っぽい) — 安全側で text/xml より image/svg+xml
    if b"<svg" in sample[:256].lower():
        return "image/svg+xml"

    # MP4 (ftyp)
    if b"ftyp" in sample[4:12]:
        return "video/mp4"
    # WebM (matroska, EBML ヘッダ 0x1A 45 DF A3)
    if sample.startswith(b"\x1A\x45\xDF\xA3"):
        # ここでは video/webm として返す
        return "video/webm"

    # MP3 (ID3) or frame sync 0xFFEx
    if sample.startswith(b"ID3") or (len(sample) > 2 and sample[0] == 0xFF and (sample[1] & 0xE0) == 0xE0):
        return "audio/mpeg"

    # WAV (RIFF....WAVE)
    if sample.startswith(b"RIFF") and b"WAVE" in sample[8:16]:
        return "audio/wav"

    # テキスト系: かなりざっくり。UTF-8/ASCII らしい時のみ text/plain に寄せる
    try:
        sample.decode("utf-8")
        # ただし HTML っぽい場合は text/html
        if b"<html" in sample[:512].lower() or b"<!doctype html" in sample[:512].lower():
            return "text/html"
        return "text/plain"
    except UnicodeDecodeError:
        pass

    return None

def _decide_content_type(bucket: str, key: str) -> str:
    """
    Content-Type の決定順序:
      1) x-amz-meta-ctype
      2) 先頭バイトのマジックナンバー
      3) S3のContentType
      4) 拡張子推定
    """
    meta_ct = None
    head_ct = None
    try:
        head = s3.head_object(Bucket=bucket, Key=key)
        head_ct = (head.get("ContentType") or "").strip() or None
        meta_ct = (head.get("Metadata", {}) or {}).get("ctype")
        if meta_ct:
            meta_ct = meta_ct.strip() or None
    except ClientError:
        pass

    if meta_ct:
        return meta_ct

    magic_ct = _sniff_by_magic(_read_magic(bucket, key))
    if magic_ct:
        return magic_ct

    if head_ct:
        return head_ct

    return _guess_content_type(key)

def _paginate_list_objects(bucket: str):
    token = None
    while True:
        kwargs = {"Bucket": bucket}
        if token:
            kwargs["ContinuationToken"] = token
        resp = s3.list_objects_v2(**kwargs)
        for obj in resp.get("Contents", []):
            yield obj
        if resp.get("IsTruncated"):
            token = resp.get("NextContinuationToken")
        else:
            break

# ---------- ルーティング ----------
def lambda_handler(event, context):
    method = event.get("httpMethod") or event.get("requestContext", {}).get("http", {}).get("method", "")
    path = event.get("path") or event.get("rawPath") or "/"
    query = event.get("queryStringParameters") or {}

    if method == "OPTIONS":
        return {"statusCode": 204, "headers": CORS_HEADERS, "body": ""}

    if not BUCKET:
        return _response(500, {"message": "環境変数 UPLOAD_BUCKET が未設定です"})

    try:
        # ---- presign（アップロード）----
        if method == "POST" and path.endswith("/presign"):
            body = {}
            if event.get("body"):
                body = json.loads(event["body"]) if not event.get("isBase64Encoded") else json.loads(
                    bytes(event["body"], "utf-8").decode("utf-8")
                )

            filename = body.get("filename") or query.get("filename")
            if not filename:
                return _response(400, {"message": "filename は必須です"})

            # クライアント申告がなくても octet-stream にフォールバック
            content_type = (body.get("contentType") or "").strip() or "application/octet-stream"

            put_url = s3.generate_presigned_url(
                "put_object",
                Params={
                    "Bucket": BUCKET,
                    "Key": filename,
                    "ContentType": content_type,            # 保存時の ContentType
                    "Metadata": {"ctype": content_type},    # 本来タイプの手掛かりをメタに保存
                },
                ExpiresIn=EXPIRES,
            )
            return _response(200, {"url": put_url, "key": filename})

        # ---- list（一覧 + 表示/ダウンロードURL）----
        if method == "GET" and path.endswith("/list"):
            items = []
            for obj in _paginate_list_objects(BUCKET):
                key = obj["Key"]
                filename = os.path.basename(key)

                ct = _decide_content_type(BUCKET, key)

                # 表示用: 安全タイプのみ inline、それ以外は attachment で強制ダウンロード
                disp_for_view = "inline" if _is_safe_inline(ct) else _disp_attachment(filename)

                view_url = s3.generate_presigned_url(
                    "get_object",
                    Params={
                        "Bucket": BUCKET,
                        "Key": key,
                        "ResponseContentDisposition": disp_for_view,
                        "ResponseContentType": ct,
                    },
                    ExpiresIn=EXPIRES,
                )

                download_url = s3.generate_presigned_url(
                    "get_object",
                    Params={
                        "Bucket": BUCKET,
                        "Key": key,
                        "ResponseContentDisposition": _disp_attachment(filename),
                        "ResponseContentType": ct,
                    },
                    ExpiresIn=EXPIRES,
                )

                items.append(
                    {
                        "key": key,
                        "size": obj.get("Size", 0),
                        "lastModified": obj.get("LastModified").isoformat() if obj.get("LastModified") else None,
                        "viewUrl": view_url,
                        "downloadUrl": download_url,
                        "contentType": ct,
                    }
                )

            return _response(200, {"items": items})

        return _response(404, {"message": "Not Found"})

    except ClientError as e:
        return _response(500, {"message": "S3エラー", "detail": str(e)})
    except Exception as e:
        return _response(500, {"message": "サーバエラー", "detail": str(e)})

# 互換: CDK の "handler": "handler.handler" 向け
def handler(event, context):
    return lambda_handler(event, context)

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

157

158

159

160

161

162

163

164

165

166

167

168

169

170

171

172

173

174

175

176

177

178

179

180

181

182

183

184

185

186

187

188

189

190

191

192

193

194

195

196

197

198

199

200

201

202

203

204

205

206

207

208

209

210

211

212

213

214

215

216

217

218

219

220

221

222

223

224

225

226

227

228

229

230

231

232

233

234

235

236

237

238

239

240

241

242

243

244

245

246

247

248

249

250

251

252

253

254

255

256

257

258

259

260

261

262

263

264

265

266

267

268

269

270

271

272

# -*- coding: utf-8 -*-

"""

API Gateway + Lambda (Python) 用のアップローダーAPIハンドラ（強化版）

変更要点:

- 「表示」リンクで拡張子が嘘でも正しく表示できるように、

Content-Type 決定ロジックを強化。

優先順位:

1) x-amz-meta-ctype（アップロード時に記録）

2) オブジェクト先頭バイトのマジックナンバー検出（PDF/PNG/JPEG/GIF/MP4/WebM/MP3/WAV/SVG）

3) S3のContentType

4) 拡張子推定（mimetypes）

- 安全に inline できるタイプのみ inline。その他は attachment。

- 日本語ファイル名のダウンロード対応（filename*）

"""

import json

import os

import re

import urllib.parse

import mimetypes

from io import BytesIO

import boto3

from botocore.exceptions import ClientError

from botocore.config import Config

# ---------- 設定 ----------

s3 = boto3.client("s3", config=Config(s3={"addressing_style": "virtual"}))

BUCKET = os.environ.get("UPLOAD_BUCKET", "")

EXPIRES = int(os.environ.get("URL_EXPIRES", "3600"))

CORS_HEADERS = {

"Content-Type": "application/json",

"Access-Control-Allow-Origin": "*",

"Access-Control-Allow-Headers": "Content-Type,Authorization",

"Access-Control-Allow-Methods": "GET,POST,OPTIONS",

}

SAFE_INLINE_PREFIXES = ("image/", "video/", "audio/", "text/")

SAFE_INLINE_EXACT = {"application/pdf"}

# ---------- ユーティリティ ----------

def _response(status: int, body) -> dict:

return {

"statusCode": status,

"headers": CORS_HEADERS,

"body": json.dumps(body, ensure_ascii=False),

}

def _ascii_fallback(name: str) -> str:

if "." in name:

stem, ext = name.rsplit(".", 1)

safe_stem = re.sub(r"[^A-Za-z0-9._-]", "_", stem) or "download"

return f"{safe_stem}.{ext}"

else:

return re.sub(r"[^A-Za-z0-9._-]", "_", name) or "download"

def _disp_attachment(filename: str) -> str:

fallback = _ascii_fallback(filename)

encoded = urllib.parse.quote(filename, safe="", encoding="utf-8")

return f'attachment; filename="{fallback}"; filename*=UTF-8\'\'{encoded}'

def _guess_content_type(key: str) -> str:

guess, _ = mimetypes.guess_type(key)

return guess or "application/octet-stream"

def _is_safe_inline(ct: str) -> bool:

return (ct in SAFE_INLINE_EXACT) or any(ct.startswith(p) for p in SAFE_INLINE_PREFIXES)

def _read_magic(bucket: str, key: str, length: int = 2048) -> bytes:

"""オブジェクト先頭を部分取得して返す。失敗時は空バイト列。"""

try:

resp = s3.get_object(Bucket=bucket, Key=key, Range=f"bytes=0-{max(0,length-1)}")

return resp["Body"].read(length)

except ClientError:

return b""

def _sniff_by_magic(sample: bytes) -> str | None:

"""先頭バイトから代表的なContent-Typeを推定。該当なしならNone。"""

if not sample:

return None

# PDF: %PDF-

if sample.startswith(b"%PDF-"):

return "application/pdf"

# PNG

if sample.startswith(b"\x89PNG\r\n\x1a\n"):

return "image/png"

# JPEG

if sample.startswith(b"\xFF\xD8\xFF"):

return "image/jpeg"

# GIF

if sample.startswith(b"GIF87a") or sample.startswith(b"GIF89a"):

return "image/gif"

# WebP (RIFF....WEBP)

if sample.startswith(b"RIFF") and b"WEBP" in sample[8:16]:

return "image/webp"

# SVG (テキストXML、先頭に <svg っぽい) — 安全側で text/xml より image/svg+xml

if b"<svg" in sample[:256].lower():

return "image/svg+xml"

# MP4 (ftyp)

if b"ftyp" in sample[4:12]:

return "video/mp4"

# WebM (matroska, EBML ヘッダ 0x1A 45 DF A3)

if sample.startswith(b"\x1A\x45\xDF\xA3"):

# ここでは video/webm として返す

return "video/webm"

# MP3 (ID3) or frame sync 0xFFEx

if sample.startswith(b"ID3") or (len(sample) > 2 and sample[0] == 0xFF and (sample[1] & 0xE0) == 0xE0):

return "audio/mpeg"

# WAV (RIFF....WAVE)

if sample.startswith(b"RIFF") and b"WAVE" in sample[8:16]:

return "audio/wav"

# テキスト系: かなりざっくり。UTF-8/ASCII らしい時のみ text/plain に寄せる

try:

sample.decode("utf-8")

# ただし HTML っぽい場合は text/html

if b"<html" in sample[:512].lower() or b"<!doctype html" in sample[:512].lower():

return "text/html"

return "text/plain"

except UnicodeDecodeError:

pass

return None

def _decide_content_type(bucket: str, key: str) -> str:

"""

Content-Type の決定順序:

1) x-amz-meta-ctype

2) 先頭バイトのマジックナンバー

3) S3のContentType

4) 拡張子推定

"""

meta_ct = None

head_ct = None

try:

head = s3.head_object(Bucket=bucket, Key=key)

head_ct = (head.get("ContentType") or "").strip() or None

meta_ct = (head.get("Metadata", {}) or {}).get("ctype")

if meta_ct:

meta_ct = meta_ct.strip() or None

except ClientError:

pass

if meta_ct:

return meta_ct

magic_ct = _sniff_by_magic(_read_magic(bucket, key))

if magic_ct:

return magic_ct

if head_ct:

return head_ct

return _guess_content_type(key)

def _paginate_list_objects(bucket: str):

token = None

while True:

kwargs = {"Bucket": bucket}

if token:

kwargs["ContinuationToken"] = token

resp = s3.list_objects_v2(**kwargs)

for obj in resp.get("Contents", []):

yield obj

if resp.get("IsTruncated"):

token = resp.get("NextContinuationToken")

else:

break

# ---------- ルーティング ----------

def lambda_handler(event, context):

method = event.get("httpMethod") or event.get("requestContext", {}).get("http", {}).get("method", "")

path = event.get("path") or event.get("rawPath") or "/"

query = event.get("queryStringParameters") or {}

if method == "OPTIONS":

return {"statusCode": 204, "headers": CORS_HEADERS, "body": ""}

if not BUCKET:

return _response(500, {"message": "環境変数 UPLOAD_BUCKET が未設定です"})

try:

# ---- presign（アップロード）----

if method == "POST" and path.endswith("/presign"):

body = {}

if event.get("body"):

body = json.loads(event["body"]) if not event.get("isBase64Encoded") else json.loads(

bytes(event["body"], "utf-8").decode("utf-8")

)

filename = body.get("filename") or query.get("filename")

if not filename:

return _response(400, {"message": "filename は必須です"})

# クライアント申告がなくても octet-stream にフォールバック

content_type = (body.get("contentType") or "").strip() or "application/octet-stream"

put_url = s3.generate_presigned_url(

"put_object",

Params={

"Bucket": BUCKET,

"Key": filename,

"ContentType": content_type, # 保存時の ContentType

"Metadata": {"ctype": content_type}, # 本来タイプの手掛かりをメタに保存

ExpiresIn=EXPIRES,

)

return _response(200, {"url": put_url, "key": filename})

# ---- list（一覧 + 表示/ダウンロードURL）----

if method == "GET" and path.endswith("/list"):

items = []

for obj in _paginate_list_objects(BUCKET):

key = obj["Key"]

filename = os.path.basename(key)

ct = _decide_content_type(BUCKET, key)

# 表示用: 安全タイプのみ inline、それ以外は attachment で強制ダウンロード

disp_for_view = "inline" if _is_safe_inline(ct) else _disp_attachment(filename)

view_url = s3.generate_presigned_url(

"get_object",

Params={

"Bucket": BUCKET,

"Key": key,

"ResponseContentDisposition": disp_for_view,

"ResponseContentType": ct,

ExpiresIn=EXPIRES,

)

download_url = s3.generate_presigned_url(

"get_object",

Params={

"Bucket": BUCKET,

"Key": key,

"ResponseContentDisposition": _disp_attachment(filename),

"ResponseContentType": ct,

ExpiresIn=EXPIRES,

)

items.append(

{

"key": key,

"size": obj.get("Size", 0),

"lastModified": obj.get("LastModified").isoformat() if obj.get("LastModified") else None,

"viewUrl": view_url,

"downloadUrl": download_url,

"contentType": ct,

}

)

return _response(200, {"items": items})

return _response(404, {"message": "Not Found"})

except ClientError as e:

return _response(500, {"message": "S3エラー", "detail": str(e)})

except Exception as e:

return _response(500, {"message": "サーバエラー", "detail": str(e)})

# 互換: CDK の "handler": "handler.handler" 向け

def handler(event, context):

return lambda_handler(event, context)

frontend/index.html

<!doctype html>
<html lang="ja">
  <head>
    <meta charset="utf-8" /> <!-- ページの文字コードをUTF-8に設定 -->
    <meta name="viewport" content="width=device-width, initial-scale=1" /> <!-- モバイル端末対応 -->
    <title>mini-cdk-uploader</title> <!-- ページタイトル -->
    <style>
      /* 全体レイアウトと見た目設定 */
      body { font-family: system-ui, sans-serif; max-width: 720px; margin: 2rem auto; padding: 0 1rem; }
      header { margin-bottom: 1rem; }
      .box { border: 1px solid #ddd; border-radius: 8px; padding: 1rem; margin-bottom: 1rem; }
      .row { display: flex; gap: .5rem; align-items: center; }
      .list { margin-top: .5rem; }
      .item { display: flex; justify-content: space-between; align-items: center; padding: .5rem 0; border-bottom: 1px solid #eee; }
      .muted { color: #666; font-size: .9em; }
      code { background: #f6f8fa; padding: .2rem .4rem; border-radius: 4px; }
      button { cursor: pointer; }
    </style>
  </head>
  <body>
    <!-- ヘッダー部 -->
    <header>
      <h1>mini-cdk-uploader</h1>
      <p>CDKで最小構成の「アップロード＋一覧＋ダウンロード」Webアプリ。</p>
      <p class="muted">※ デプロイ後に <code>API_BASE_URL</code> を下のスクリプト内で置換してください（CDK出力参照）。</p>
    </header>

    <!-- アップロードセクション -->
    <section class="box">
      <h2>1) アップロード</h2>
      <div class="row">
        <input type="file" id="file" /> <!-- ファイル選択ボタン -->
        <button id="btn-upload">アップロード</button> <!-- アップロード開始ボタン -->
      </div>
      <p id="status" class="muted"></p> <!-- ステータスメッセージ表示 -->
    </section>

    <!-- 一覧＆ダウンロードセクション -->
    <section class="box">
      <h2>2) 一覧＆ダウンロード</h2>
      <button id="btn-reload">再読込</button> <!-- 一覧再取得ボタン -->
      <div id="list" class="list"></div> <!-- ファイル一覧表示領域 -->
    </section>

    <script>
      // ▼▼▼ デプロイ後のAPI GatewayのURLをここに指定
      const API_BASE_URL = 'https://xxxxxx.execute-api.ap-northeast-1.amazonaws.com/prod/';

      // DOM取得のショートカット関数
      const $ = (sel) => document.querySelector(sel);

      // presign URLを取得し、S3にアップロードする関数
      async function presignAndUpload(file) {
        // Lambda(API Gateway経由)にリクエストを送りpresigned URLを取得
        const res = await fetch(`${API_BASE_URL}presign`, {
          method: 'POST',
          headers: { 'Content-Type': 'application/json' },
          body: JSON.stringify({ filename: file.name, contentType: file.type || 'application/octet-stream' })
        });
        if (!res.ok) throw new Error('presign失敗'); // 失敗時
        const data = await res.json();

        // 取得したURLにPUTでファイルアップロード
        const put = await fetch(data.url, {
          method: 'PUT',
          mode: 'cors',            // CORSを明示的に許可
          credentials: 'omit',     // cookieなどの資格情報を送信しない
          headers: { 'Content-Type': file.type || 'application/octet-stream' },
          body: file,              // アップロードするファイル
        });
        if (!put.ok) throw new Error('アップロード失敗'); // アップロード失敗時
      }

      // バケット内のファイル一覧を取得して表示する関数
      async function loadList() {
        const list = $('#list');
        list.textContent = '読込中...';
        const res = await fetch(`${API_BASE_URL}list`); // Lambda経由で一覧取得
        if (!res.ok) {
          list.textContent = '一覧取得に失敗しました';
          return;
        }
        const data = await res.json();
        const items = data.items || [];
        if (!items.length) {
          list.textContent = 'ファイルはまだありません';
          return;
        }
        const frag = document.createDocumentFragment(); // DOMフラグメント作成（高速に構築）

        // 各ファイルを行として描画
        items.forEach(it => {
          const row = document.createElement('div');
          row.className = 'item';
          const left = document.createElement('div');
          // ファイル名、サイズ、更新日、Content-Type表示
          left.innerHTML = `<strong>${it.key}</strong><br><span class="muted">${(it.size||0)} bytes / ${it.lastModified||''} / ${it.contentType||''}</span>`;
          const right = document.createElement('div');

          // 表示リンク（別タブで開く）
          const view = document.createElement('a');
          view.href = it.viewUrl;
          view.textContent = '表示';
          view.target = '_blank'; // 新しいタブで開く
          view.rel = 'noopener';  // セキュリティ対策

          // 区切り文字
          const sep = document.createTextNode(' / ');

          // ダウンロードリンク（attachment設定済み）
          const dl = document.createElement('a');
          dl.href = it.downloadUrl;
          dl.textContent = 'ダウンロード';

          // 要素を組み立て
          right.appendChild(view);
          right.appendChild(sep);
          right.appendChild(dl);
          row.appendChild(left);
          row.appendChild(right);
          frag.appendChild(row);
        });

        // 一覧を更新
        list.innerHTML = '';
        list.appendChild(frag);
      }

      // アップロードボタン押下時の処理
      $('#btn-upload').addEventListener('click', async () => {
        const file = $('#file').files[0]; // ファイルを取得
        if (!file) {
          $('#status').textContent = 'ファイルを選択してください';
          return;
        }
        $('#status').textContent = 'アップロード中...';
        try {
          await presignAndUpload(file); // アップロード処理実行
          $('#status').textContent = 'アップロード完了！';
          await loadList(); // アップロード後に一覧更新
        } catch (e) {
          console.error(e);
          $('#status').textContent = 'エラー: ' + e.message; // エラー表示
        }
      });

      // 再読込ボタン押下時の処理
      $('#btn-reload').addEventListener('click', loadList);

      // ページ読み込み時に自動で一覧を取得（API未設定時は警告）
      (async()=>{try{await loadList();}catch(e){console.warn('API未設定');}})();
    </script>
  </body>
</html>

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

<!doctype html>

<head>

<title>mini-cdk-uploader</title>

<style>

/* 全体レイアウトと見た目設定 */

body { font-family: system-ui, sans-serif; max-width: 720px; margin: 2rem auto; padding: 0 1rem; }

header { margin-bottom: 1rem; }

.box { border: 1px solid #ddd; border-radius: 8px; padding: 1rem; margin-bottom: 1rem; }

.row { display: flex; gap: .5rem; align-items: center; }

.list { margin-top: .5rem; }

.item { display: flex; justify-content: space-between; align-items: center; padding: .5rem 0; border-bottom: 1px solid #eee; }

.muted { color: #666; font-size: .9em; }

code { background: #f6f8fa; padding: .2rem .4rem; border-radius: 4px; }

button { cursor: pointer; }

</style>

</head>

<body>

<h1>mini-cdk-uploader</h1>

<p>CDKで最小構成の「アップロード＋一覧＋ダウンロード」Webアプリ。</p>

<p class="muted">※ デプロイ後に <code>API_BASE_URL</code> を下のスクリプト内で置換してください（CDK出力参照）。</p>

</header>

<h2>1) アップロード</h2>

<button id="btn-upload">アップロード</button>

</div>

</section>

<h2>2) 一覧＆ダウンロード</h2>

</section>

// ▼▼▼ デプロイ後のAPI GatewayのURLをここに指定

const API_BASE_URL = 'https://xxxxxx.execute-api.ap-northeast-1.amazonaws.com/prod/';

// DOM取得のショートカット関数

const $ = (sel) => document.querySelector(sel);

// presign URLを取得し、S3にアップロードする関数

async function presignAndUpload(file) {

// Lambda(API Gateway経由)にリクエストを送りpresigned URLを取得

const res = await fetch(`${API_BASE_URL}presign`, {

method: 'POST',

headers: { 'Content-Type': 'application/json' },

body: JSON.stringify({ filename: file.name, contentType: file.type || 'application/octet-stream' })

});

if (!res.ok) throw new Error('presign失敗'); // 失敗時

const data = await res.json();

// 取得したURLにPUTでファイルアップロード

const put = await fetch(data.url, {

method: 'PUT',

mode: 'cors', // CORSを明示的に許可

credentials: 'omit', // cookieなどの資格情報を送信しない

headers: { 'Content-Type': file.type || 'application/octet-stream' },

body: file, // アップロードするファイル

});

if (!put.ok) throw new Error('アップロード失敗'); // アップロード失敗時

}

// バケット内のファイル一覧を取得して表示する関数

async function loadList() {

const list = $('#list');

list.textContent = '読込中...';

const res = await fetch(`${API_BASE_URL}list`); // Lambda経由で一覧取得

if (!res.ok) {

list.textContent = '一覧取得に失敗しました';

return;

}

const data = await res.json();

const items = data.items || [];

if (!items.length) {

list.textContent = 'ファイルはまだありません';

return;

}

const frag = document.createDocumentFragment(); // DOMフラグメント作成（高速に構築）

// 各ファイルを行として描画

items.forEach(it => {

const row = document.createElement('div');

row.className = 'item';

const left = document.createElement('div');

// ファイル名、サイズ、更新日、Content-Type表示

left.innerHTML = `<strong>${it.key}</strong><br><span class="muted">${(it.size||0)} bytes / ${it.lastModified||''} / ${it.contentType||''}</span>`;

const right = document.createElement('div');

// 表示リンク（別タブで開く）

const view = document.createElement('a');

view.href = it.viewUrl;

view.textContent = '表示';

view.target = '_blank'; // 新しいタブで開く

view.rel = 'noopener'; // セキュリティ対策

// 区切り文字

const sep = document.createTextNode(' / ');

// ダウンロードリンク（attachment設定済み）

const dl = document.createElement('a');

dl.href = it.downloadUrl;

dl.textContent = 'ダウンロード';

// 要素を組み立て

right.appendChild(view);

right.appendChild(sep);

right.appendChild(dl);

row.appendChild(left);

row.appendChild(right);

frag.appendChild(row);

});

// 一覧を更新

list.innerHTML = '';

list.appendChild(frag);

}

// アップロードボタン押下時の処理

$('#btn-upload').addEventListener('click', async () => {

const file = $('#file').files[0]; // ファイルを取得

if (!file) {

$('#status').textContent = 'ファイルを選択してください';

return;

}

$('#status').textContent = 'アップロード中...';

try {

await presignAndUpload(file); // アップロード処理実行

$('#status').textContent = 'アップロード完了！';

await loadList(); // アップロード後に一覧更新

} catch (e) {

console.error(e);

$('#status').textContent = 'エラー: ' + e.message; // エラー表示

}

});

// 再読込ボタン押下時の処理

$('#btn-reload').addEventListener('click', loadList);

// ページ読み込み時に自動で一覧を取得（API未設定時は警告）

(async()=>{try{await loadList();}catch(e){console.warn('API未設定');}})();

</script>

</body>

</html>

関連記事