투케이2K

220. (python/파이썬) [AWS] [Lambda] 런타임 Python 3.13 - boto3 모듈 사용해 Long URL 형식 S3 GetPresignedUrl 확인 본문

Python

220. (python/파이썬) [AWS] [Lambda] 런타임 Python 3.13 - boto3 모듈 사용해 Long URL 형식 S3 GetPresignedUrl 확인

투케이2K 2025. 12. 19. 19:22
728x90

[개발 환경 설정]

개발 툴 : Aws / Lambda / Runtime Python 3.13

개발 언어 : python

 

[소스 코드]

// --------------------------------------------------------------------------------------
[개발 및 테스트 환경]
// --------------------------------------------------------------------------------------

- 언어 : Python


- 개발 툴 : Aws / Lambda / Runtime Python 3.13


- 개발 기술 : AWS Lambda 이벤트 동작 함수


- 사전) AWS Lambda 설명 : 

  >> Aws Lambda 는 서버 리스 FaaS 솔루션으로, 함수의 인스턴스를 실행하여 이벤트를 처리할 수 있습니다

  >> Aws Lambda 는 이벤트에 응답하여 코드를 실행 하고 해당 코드에 필요한 컴퓨팅 리소스를 자동으로 관리합니다


- 사전) AWS S3 간략 설명 : 

  >> Aws S3 버킷 이란 데이터 (사진, 동영상, 문서 등) 객체 를 업로드할 수 있는 컨테이너 (디렉토리) 입니다

  >> Aws S3 버킷은 온라인 스토리지 서비스로 HTTP/HTTPS 를 통한 API 를 사용해 파일 업로드 및 다운로드 처리를 할 수 있습니다


- 사전) STS 임시 보안 자격 증명 설명 정리 : 

  >> AWS STS 는 AWS 리소스에 대한 액세스를 제어할 수 있는 임시 보안 자격 증명입니다 (신뢰받는 사용자에게 제공)

  >> AWS STS 는 단기적 임시 보안 자격 증명이며, 몇 분에서 몇 시간 동안 해당 자격 증명을 사용해 AWS 리소스를 액세스할 수 있습니다

  >> AWS STS 임시 보안 자격 증명이 만료 된 경우 AWS는 더는 그 자격 증명을 인식하지 못하거나 그 자격 증명을 사용한 API 요청으로부터 이루어지는 어떤 종류의 액세스도 허용하지 않습니다 

// --------------------------------------------------------------------------------------






// --------------------------------------------------------------------------------------
[소스 코드]
// --------------------------------------------------------------------------------------

# ========================================================================
# [Aws] : [boto3 모듈] : IAM 계정 정보 사용해 Aws STS 임시 자격 증명 조회 및 S3 Get PresignedUrl 생성
# ========================================================================
# [참고] : API Gateway 와 연동 되어 Post 방식으로 Lambda 함수 호출 및 응답 값 반환 수행
# ========================================================================
"""
1. aws lambda python 3.13 런타임 환경 기반
2. boto3 모듈 기본 내장 AWS 사용 모듈
3. import ClientError : AWS SDK for Python 인 boto3 에서 발생할 수 있는 예외 중 하나로, AWS 서비스 호출 중 오류가 발생했을 때 사용됩니다
"""
# ========================================================================

import json

import os
import urllib.parse

import random
import string

import boto3
from botocore.exceptions import ClientError
from decimal import Decimal


def random_string(length): # 알파벳 + 숫자 랜덤 문자열 생성 함수
    chars = string.ascii_letters + string.digits  # A-Z, a-z, 0-9
    return ''.join(random.choice(chars) for _ in range(length))


def lambda_handler(event, context): # Lambda 호출 시 동작 되는 메인 함수

    # [event , context 정보 디버깅 로그 출력]
    print(f"DLOG = event : {event} / context {context}")


    # [Return 반환 Json 변수 선언]
    response = {
        "statusCode" : 0,
        "headers" : {},
        "body" : ""
    }


    # [AWS IAM 계정 AccessKey, SecretKey 변수 선언]
    iamAccessKey = "AK..7Q"
    iamSecretKey = "Zz..xj"
    iamRegion = "ap-northeast-2"  # 리전


    # [명시적 인증 정보로 세션 생성]
    session = boto3.Session(
        aws_access_key_id = iamAccessKey,
        aws_secret_access_key = iamSecretKey,
        region_name = iamRegion
    )


    # [STS 클라이언트 생성]
    sts_client = session.client('sts')


    # [현재 인증 정보 확인]    
    identity = sts_client.get_caller_identity()

    print(f"DLOG = identity : {identity}")

    
    try:

      # [STS 임시 자격 증명 및 S3 프리사인 URL 유효 기간 설정]
      durationSeconds = 36000 # ✅ 3600 은 1시간 / 36000 은 10 시간


      # [STS 정보 확인 수행]
      stsRes = sts_client.get_session_token(DurationSeconds=durationSeconds) # ✅ durationSeconds 지정

      credentials = stsRes['Credentials']
      
      print(f"DLOG = credentials : {credentials}")

      stsAccessKey = credentials['AccessKeyId']
      stsSecretKey = credentials['SecretAccessKey']
      stsSessionToken = credentials['SessionToken']


      # [S3 클라이언트 생성]
      s3_client = boto3.client( # ✅ 새로운 클라이언트 생성
          's3',        
          aws_access_key_id=stsAccessKey,
          aws_secret_access_key=stsSecretKey,
          aws_session_token=stsSessionToken,
          region_name = iamRegion # 리전 정보
      )


      # [S3 버킷 저장소 정보 지정]
      s3DefaultFileName = "work_private.txt"
      s3BucketName = "service"
      s3BucketPrefix = f"control/{s3DefaultFileName}"


      # [S3 Get PresignedUrl 호출 파라미터 생성]
      #downloadFileName = 'adbdsdfsafdsafaf' + '-' + s3DefaultFileName # ✅ 다운로드 될 파일 명칭 생성 Long 형식
      downloadFileName = random_string(100) + '-' + s3DefaultFileName # ✅ 다운로드 될 파일 명칭 생성 Long 형식
      
      print(f"DLOG = downloadFileName : {downloadFileName}")

      s3Params={
        'Bucket': s3BucketName, 
        'Key': s3BucketPrefix,
        'ResponseContentDisposition': f'attachment; filename="{downloadFileName}"'
      }


      # [S3 Get PresignedUrl 정보 확인 수행]
      s3Res = s3_client.generate_presigned_url(
          'get_object', # ✅ 파일 다운로드 명시
          Params=s3Params,
          ExpiresIn=durationSeconds # ✅ durationSeconds 지정
      )

      print(f"DLOG = s3Res : {s3Res}")


      # [리턴 변수 삽입] : ApiGateWay 응답 반환 설정 : Lambda 통합 요청 사용
      response["statusCode"] = 200 
      response["headers"] = {
        "Content-Type": "application/json",
        'Access-Control-Allow-Origin': '*',  # CORS 허용 (필요 시)
      }
      response["body"] = json.dumps({
        "s3Res": s3Res
      })
        
    except ClientError as e: # AWS 서비스 호출 중 오류 발생 처리
        error_code = e.response['Error']['Code']
        error_message = e.response['Error']['Message']

        # [리턴 변수 삽입]
        response["statusCode"] = 400
        response["headers"] = {
          "Content-Type": "application/json",
          'Access-Control-Allow-Origin': '*',  # CORS 허용 (필요 시)
        }
        response["body"] = json.dumps(
          {
            "exception" : "ClientError",
            "error_code" : error_code,
            "error_message" : error_message
          }
        )


    # [리턴 반환 수행]
    return response

# ========================================================================
"""
Response:
{
  "statusCode": 200,
  "headers": {
    "Content-Type": "application/json",
    "Access-Control-Allow-Origin": "*"
  },
  "body": "{\"s3Res\": \"https://service.s3.amazonaws.com/control/work_private.txt?response-content-disposition=attachment%3B%20filename%3D%22L1Y9H44JmdLuRnLbL0KLEusZ2P85AmqGE8u4qgp7Be2Ji1xSHNC84LFcRVW4IDgRsEQbOUWaRieOGYP5cJcIcI5yMhpALizSl3F1-work_private.txt%22&X-Amz-Algorithm=AWS4-HMAC-SHA256&X-Amz-Credential=AS..OB%2F20251218%2Fap-northeast-2%2Fs3%2Faws4_request&X-Amz-Date=20251218T002219Z&X-Amz-Expires=36000&X-Amz-SignedHeaders=host&X-Amz-Security-Token=IQoJ..Bll0%2FoKK2rTJTVTnpe3aSB3Usqg%3D%3D&X-Amz-Signature=749..09\"}"
}
"""
# ========================================================================

// --------------------------------------------------------------------------------------






// --------------------------------------------------------------------------------------
[참고 사이트]
// --------------------------------------------------------------------------------------

[자바스크립트 AWS STS 임시 자격 증명 사용해 S3 Get PreSignedUrl 프리 사인 URL 주소 생성]

https://kkh0977.tistory.com/8151

https://blog.naver.com/kkh0977/223938740405


[Aws S3 Storage] PreSignedUrl 프리 사인 URL 주소 정리 - S3 버킷 저장소 Get 확인 및 Put 업로드 임시 권한 주소

https://blog.naver.com/kkh0977/223903771897


[Aws S3 Storage] S3 (Amazon Simple Storage Service) 버킷 저장소 개념 및 설명 정리

https://blog.naver.com/kkh0977/223733087281?trackingCode=blog_bloghome_searchlist


[Amazon API Gateway] Aws API Gateway 게이트웨이 설명 정리 - 중개 서버

https://blog.naver.com/kkh0977/223827753479


[Amazon API Gateway] Aws API Gateway 게이트웨이 API 엔드포인트 유형 정리

https://blog.naver.com/kkh0977/223911565693


[Aws Lambda] Aws 사이트에서 생성 된 Lambda 람다 검증 함수 리스트 및 내용 소스 코드 확인 방법

https://blog.naver.com/kkh0977/223765198383


[AWS] Lambda 람다 함수 수행 errorType Sandbox.Timedout 에러 발생

https://blog.naver.com/kkh0977/223962778768

// --------------------------------------------------------------------------------------
 
728x90
반응형
Comments